我正在并行化一个高度依赖WinAPI NetAPI32调用的方法。如果用户输入的主机已关闭或数百个主机列表中的多个主机,则调用有时会超时。
int prevThreads, prevPorts;
ThreadPool.GetMinThreads(out prevThreads, out prevPorts);
ThreadPool.SetMinThreads(20, prevPorts);
var parallelScanList = computersToScan.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism).WithDegreeOfParallelism(20);
Api.WinApi.AdvApi.LogonAndImpersonate(connection.UserCredential);
foreach (var computer in parallelScanList)
{
//...
//this takes a long time to timeout
status = NetApi.NetUserEnum(computer.DnsHostname, 2,
(int)NetApi.NetUserEnumFilter.FILTER_NORMAL_ACCOUNT,
out userbufPtr, (int)LmCons.MAX_PREFERRED_LENGTH, out userEntriesRead, out totalEntries,
out userResumeHandle);
}
我们在使用消费者/生产者的 C 客户端中具有与此类似的逻辑。 启动 20 个线程并让它们读取列表,直到它耗尽。
function StartProcessingHosts()
{
for 1 to 20
StartProcessThread()
}
function ProcessHostsThread()
{
while(moreHosts)
{
//obviously synchronization around here
var host = popHost();
DoSomething(host);
}
}
这非常快,因为这些网络调用的所有等待都在进行,并且可能无法连接到被关闭的主机。
我目前在 C# 中执行此操作的方式似乎是一次一个。
更新:
我明白了,问题是foreach循环。您可能已经假设,通过使查询AsParallel
然后在 foreach 中执行它将使其并行。当然,这不会发生。此外,使用 PLINQ,您可以实现与 svick 答案中演示的相同目标。
但是,这是并行化代码的另一种方法,我在下面提到它,因为 svick 的答案也受到这样一个事实的影响,即仅通过设置MaxDegreeOfParallelism = 20
并不能保证 20 次并行执行。它仍然只是并行执行的上限,而不是下限。如果 PLINQ 执行引擎认为它应该只启动 5 个并行执行,它将只启动 5 个并行执行,这是完全合法的执行。
以下代码保证 20 次并行执行:
var concurrentScanList = new ConcurrentQueue<Computer>(computersToScan);
var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None);
var taskArray = new Task[20];
//Initializing the tasks
for (var index = 0; index < taskArray.Length; index++)
{
taskArray[index] = taskFactory.StartNew(() =>
{
Computer host;
while (concurrentScanList.TryDequeue(out host))
{
DoSomething(host);
}
});
}
//Wait for all tasks to finish - queue will be empty then
Task.WaitAll(baseProcessorTaskArray);
旧答案:
WithDegreeOfParallelism()
是,
将用于处理查询的并发执行任务的最大数量。
。我认为,由于并发执行任务的最小数量不是固定的,因此可能是 1。
从本质上讲,您的猜测可能是正确的,即此执行不是并行发生的,因此超时。此外,即使它以等于 20 的并行度并行发生,也不能保证始终如此。
我的建议是,您将"要扫描的计算机"放在BlockingCollection中,然后生成20个任务,每个任务从该BlockingCollection中读取一台计算机,然后对其进行扫描。此实现自然是生产者使用者,因为这是问题设计的内在品质。
PLINQ,Parallel LINQ 的缩写,你猜对了,用于并行化 LINQ 查询。例如,如果编写 collection.AsParallel().Where(/* some condition */).Select(/* some projection */).ToList()
,则Where()
和Select()
将并行执行。
但是你不这样做,你调用AsParallel()
,说"以下LINQ查询应该并行执行"。然后,通过调用 WithExecutionMode()
和 WithDegreeOfParallelism()
来配置即将到来的查询的并行度。然后,您实际上没有任何 LINQ 查询,而是使用 foreach
,这将串行迭代集合。
如果要并行执行foreach
,则不需要 PLINQ,而是需要Parallel.ForEach()
:
Parallel.ForEach(computersToScan, new ParallelOptions { MaxDegreeOfParallelism = 20 },
computer =>
{
//...
});