如何在 bash 脚本中实现对基于文件的队列的多线程访问

Question

我在设计一个多处理 bash 脚本时遇到了一些问题，该脚本通过网站、跟踪找到的链接并在每个新页面上进行一些处理（它实际上收集电子邮件地址，但这是一个不重要的问题细节）。

脚本应该是这样工作的：

1. 下载页面
2. 解析所有链接并将它们添加到队列中
3. 做一些不重要的处理
4. 从队列中弹出一个 URL 并开始于

这本身很容易编程，问题来自两个限制和脚本需要具有的功能。

• 脚本不得两次处理一个 URL
• 脚本必须能够同时处理 n 个（作为参数提供）页面
• 脚本必须符合 POSIX（curl 除外）-> 所以没有花哨的锁

现在，我设法提出了一个实现，它使用两个文件作为队列，其中一个存储所有已处理的 URL，另一个存储已找到但尚未处理的 URL。

主进程简单地生成一堆共享队列文件的子进程，并且（在一个循环中，直到 URLs-to-be-processed-queue 为空）从URLs-to-be-processed-queue 弹出一个 URL，处理页面，尝试将每个新找到的链接添加到URLs-already-processed-queue，如果成功（URL 不存在）也将其添加到URLs-to-be-processed-queue。

问题在于您不能（AFAIK）使队列文件操作原子化，因此锁定是必要的。并且以 POSIX 兼容的方式锁定是......恐怖......缓慢的恐怖。

我的做法如下：

#Pops first element from a file ($1) and prints it to stdout; if file emepty print out empty return 1
fuPop(){
if [ -s "$1" ]; then
        sed -nr '1p' "$1"
        sed -ir '1d' "$1"
        return 0
else
        return 1
fi
}

#Appends line ($1) to a file ($2) and return 0 if it's not in it yet; if it, is just return 1
fuAppend(){
if grep -Fxq "$1" < "$2"; then
        return 1
else
        echo "$1" >> "$2"
        return 0
fi
}

#There're multiple processes running this function.
prcsPages(){
while [ -s "$todoLinks" ]; do
        luAckLock "$linksLock"
        linkToProcess="$(fuPop "$todoLinks")"
        luUnlock "$linksLock"

        prcsPage "$linkToProcess"
        ...
done
...
}


#The prcsPage downloads it, does some magic and than calls prcsNewLinks and prcsNewEmails that both get list of new emails / new urls in $1
#$doneEmails, ..., contain file path, $mailLock, ..., contain dir path

prcsNewEmails(){
luAckLock "$mailsLock"
for newEmail in $1; do
        if fuAppend "$newEmail" "$doneEmails"; then
                echo "$newEmail"
        fi
done
luUnlock "$mailsLock"
}

prcsNewLinks(){
luAckLock "$linksLock"
for newLink in $1; do
        if fuAppend "$newLink" "$doneLinks"; then
                fuAppend "$newLink" "$todoLinks"
        fi
done
luUnlock "$linksLock"
}

问题是我的实现很慢（就像真的很慢），几乎慢到使用超过 2 10 （减少锁等待有很大帮助）孩子没有意义过程。您实际上可以禁用锁（只需注释掉 luAckLock 和 luUnlock 位），它工作得很好（而且速度更快），但是seds -i 每隔一段时间就会出现竞争条件，感觉不对.

最糟糕的（如我所见）锁定prcsNewLinks，因为它需要很长时间（基本上大部分时间运行）并且实际上阻止其他进程开始处理新页面（因为它需要弹出新页面来自（当前锁定）$todoLinks 队列的 URL）。

现在我的问题是，如何做得更好、更快、更好？

整个脚本是here（它包含一些信号魔法、大量调试输出，而且通常不是那么好的代码）。

顺便说一句：是的，你是对的，在 bash 中执行此操作 - 更符合 POSIX 的方式 - 太疯狂了！但这是大学作业，所以我不得不这样做

//虽然我觉得我并没有真正期望我解决这些问题（因为竞争条件只有在拥有 25 个以上线程时才会更频繁地出现，这可能不是一个理智的人会测试的）。

---

代码注释：

1. 是的，等待应该有（并且已经有）一个随机时间。此处分享的代码只是在实际分析讲座中编写的概念证明。
2. 是的，调试输出的数量及其格式非常糟糕，应该有独立的日志记录功能。不过，这不是我的问题所在。

Answer 1

首先，您需要实现自己的 HTML/HTTP 爬虫吗？为什么不让wget 或curl 为您在站点中递归？

您可以滥用文件系统作为数据库，并让您的队列成为单行文件的目录。 （或文件名是数据的空文件）。这将为您提供生产者-消费者锁定，生产者接触文件，消费者将其从传入目录移动到处理/完成目录。

这样做的美妙之处在于，多个线程接触同一个文件即可。所需的结果是 url 在“传入”列表中出现一次，这就是当多个线程创建具有相同名称的文件时得到的结果。由于您需要重复数据删除，因此在写入传入列表时不需要锁定。

1) 下载页面

2) 解析出所有链接并将它们添加到队列中

对于找到的每个链接，

grep -ql "$url" already_checked || : > "incoming/$url"

或

[[ -e "done/$url" ]] || : > "incoming/$url"

3) 做一些不重要的处理

4) 从队列中弹出一个 URL 并从 1) 开始

# mostly untested, you might have to debug / tweak this
local inc=( incoming/* )
# edit: this can make threads exit sooner than desired.
# See the comments for some ideas on how to make threads wait for new work
while [[ $inc != "incoming/*" ]]; do
    # $inc is shorthand for "${inc[0]}", the first array entry
    mv "$inc" "done/" || { rm -f "$inc"; continue; } # another thread got that link, or that url already exists in done/
    url=${inc#incoming/}
    download "$url";
    for newurl in $(link_scan "$url"); do
        [[ -e "done/$newurl" ]] || : > "incoming/$newurl"
    done
    process "$url"
    inc=( incoming/* )
done

编辑：将 URL 编码为不包含 / 的字符串作为练习留给读者。虽然 urlencoding / 到 %2F 可能会工作得很好。

我正在考虑将 URL 移动到每个线程的“处理”列表中，但实际上如果您不需要能够从中断中恢复，您的“完成”列表可以改为“排队和完成”列表。我认为它实际上对 mv "$url" "threadqueue.$$/" 或其他什么有用。

“done/”目录会变得非常大，并且开始变慢，可能有 10k 个文件，具体取决于您使用的文件系统。将“完成”列表维护为每行一个 url 的文件或数据库（如果有一个对单个命令快速的数据库 CLI 界面）可能更有效。

将完成的列表作为文件进行维护也不错，因为您永远不需要删除条目。即使有多个附加它的进程，您也可以在不锁定它的情况下逃脱。 （我不确定如果线程 B 在线程 A 打开文件和线程 A 进行写入之间的 EOF 处写入数据会发生什么。线程 A 的文件位置可能是旧的 EOF，在这种情况下它会覆盖线程 B 的条目，或者更糟, 只覆盖它的一部分。如果你确实需要锁定，也许flock(1) 会很有用。它会获得一个锁定，然后执行你作为 args 传递的命令。）

如果没有因缺少写锁定而损坏的文件，那么您可能不需要写锁定。与每次检查/追加都必须锁定相比，“完成”列表中偶尔出现的重复条目将是一个微小的减速。

如果您需要严格正确地避免多次下载同一个 URL，则需要读者等待作者完成。如果您可以在最后sort -u 电子邮件列表，那么在附加列表时阅读旧副本对于读者来说并不是一场灾难。然后写入者只需要相互锁定，读取者就可以读取文件。如果他们在作家设法编写新条目之前达到 EOF，那就这样吧。

我不确定线程​​在从传入列表中删除条目之前或之后将条目添加到“完成”列表是否重要，只要他们在处理之前将它们添加到“完成”。我在想，一种或另一种方式可能会使比赛更有可能导致重复的完成条目，并且不太可能进行重复的下载/处理，但我不确定。