如何制定此多线程编程方案
本文关键字:编程 方案 多线程 何制定 | 更新日期: 2023-09-27 18:36:08
>我正在尝试使用多线程来遍历树结构。这里的问题是,如果没有HTTP调用,树的结构是未知的(即,HTTP调用将为您提供节点的子级)。因此,我正在尝试使用多线程来增加我们可以发出的 HTTP 请求的吞吐量。
我不知道我们应该如何很好地解决这个问题,所以我先在这里描述我的想法。
最初,我认为它类似于我们通常在BFS中编写的内容,假设我们的并发级别为10。
SemaphoreSlim semaphore = new SemaphoreSlim(10);
Task HTTPGet(Node node) {
blah blah
Q.push(childNodes);
}
while (!Q.isEmpty()) {
Node node = Q.head();
Q.pop();
taskList.Add(Task.Start(() => HTTPGet(node));
}
这里的问题是:处理第一个节点后,Q 变为空,整个循环将被终止。所以我觉得我们也需要检查信号量的剩余计数。因此,如果信号量的剩余计数不是 10,则表示某个进程仍在工作,我们应该等待它的进程。
while (!Q.isEmpty() || semaphore.Count != 10) {
Node node = Q.head();
Q.pop();
taskList.Add(Task.Start(() => HTTPGet(node));
}
但显然在第一个节点被弹出后,Q 仍然是空的,我们需要在 while 循环中做一些"等待"以确保我们可以得到节点。
while (!Q.isEmpty() || semaphore.Count != 10) {
if (Q.isEmpty()) {
Wait till Q becomes non empty
or semaphore.Count == 10 again
}
Node node = Q.head();
Q.pop();
taskList.Add(Task.Start(() => HTTPGet(node));
}
但是这变得如此丑陋,我很确定应该有更好的方法来解决这个问题。我试图在生产者-消费者范式中制定它,但失败了(因为这个时候消费者也将启动更多的生产者)。
有没有更好的方法来解决这个问题?
通过代码解释更容易,但请注意,这不是我尝试或测试过的东西。这是为了让您在正确的道路上开始
class Program {
static void Main(string[] args) {
new Program();
}
Program() {
Node root = new Node("root");
root.Children = new Node[2];
root.Children[0] = new Node("child0");
root.Children[1] = new Node("child1");
MultiThreadedBFS(root);
}
BlockingCollection<Node> Queue = new BlockingCollection<Node>(10); // Limit it to the number of threads
Node[] HTTPGet(Node parentNode) {
return parentNode.Children; //your logic to fetch nodes go here
}
volatile int ThreadCount;
void MultiThreadedBFS(Node root) {
Queue.Add(root);
// we fetch each node's children on a separate thread.
// This means that when all nodes are fetched, there are
// no more threads left. That will be our exit criteria
ThreadCount = 0;
do {
var node = FetchNextNode();
if (node == null)
break;
ProcessNode(node);
} while (true);
}
Node FetchNextNode() {
Node node;
while (!Queue.TryTake(out node, 100)) {
if (ThreadCount == 0 && Queue.Count == 0)
return null; // All nodes have been fetched now
}
return node;
}
void ProcessNode(Node node) {
// you can use a threadpool or task here
new Thread(() => {
Thread.CurrentThread.Name = "ChildThread";
++ThreadCount;
Debug.WriteLine("Retrieving children for Node: " + node);
var children = HTTPGet(node);
foreach (var child in children) {
Debug.WriteLine("Adding node for further processing: " + node);
while (!Queue.TryAdd(child, -1))
;
}
--ThreadCount;
}).Start();
}
// this is the actual node class that represents the Node on the tree
[DebuggerDisplay("Name = {Name}")]
class Node {
public string Name;
public Node[] Children = new Node[0];
public Node(string name) {
Name = name;
}
public override string ToString() {
return Name;
}
}
}
编辑:
我现在更新了程序以修复退出条件和其他一些错误
另外,即使我在这里使用线程,我认为这是使用 async/await 的完美案例。我会让其他人使用 async/await 回答
在从队列中获取之前,使每个线程立即从队列中提取的线程计数器递增。如果该计数器恰好达到线程数,则意味着所有线程都在尝试取消排队,并且不可能有任何正在进行的操作。在这种情况下,会发出所有线程退出的信号。