我正在调用一个工作方法,该方法调用数据库,然后迭代并返回值以进行并行处理。为了防止它对数据库造成负担,我在其中使用了Thread.Sleep来暂停对数据库的执行。然而,这似乎会阻塞仍在进行的Parallel.ForEach执行。如何最好地实现这一点以防止阻塞?
private void ProcessWorkItems()
{
_cancellation = new CancellationTokenSource();
_cancellation.Token.Register(() => WorkItemRepository.ResetAbandonedWorkItems());
Task.Factory.StartNew(() =>
Parallel.ForEach(GetWorkItems().AsParallel().WithDegreeOfParallelism(10), workItem =>
{
var x = ItemFactory(workItem);
x.doWork();
}), _cancellation.Token);
}
private IEnumerable<IAnalysisServiceWorkItem> GetWorkItems()
{
while (!_cancellation.IsCancellationRequested)
{
var workItems = WorkItemRepository.GetItemList(); //database call
workItems.ForEach(item =>
{
item.QueueWorkItem(WorkItemRepository);
});
foreach (var item in workItems)
{
yield return item;
}
if (workItems.Count == 0)
{
Thread.Sleep(30000); //sleep this thread for 30 seconds if no work items.
}
}
yield break;
}
编辑: 我修改了代码以包含答案,但它仍未按照我的期望工作。我在GetWorkItems()调用中添加了.AsParallel().WithDegreeOfParallelism(10)。当基础线程处于休眠状态时,我认为并行应该继续执行,这个想法是错误的吗?
例子: 我有15个项目,它遍历并获取10个项目并开始处理。每当一个项目完成时,它会从GetWorkItems请求另一个项目,直到尝试请求第16个项目。此时,它应该停止尝试获取更多项目,但应继续处理11-15号项目,直到这些项目完成。并行应该是这样工作的吗?因为它目前并没有这样做。目前正在完成6个项目时,它会锁定在Parallel.ForEach中仍在运行的后续10个项目。
WorkItemRepository.GetItemList
会执行类似于SELECT * FROM workqueue WHERE status = 'unprocessed'
的操作,并将其编译成可排队的工作项类型。 - Bryan Boettcher