标签: parallel.foreach

我在上面写了一个库,System.Data.SQLite并意识到它(我的库)在使用时有不可预知的行为Parallel.ForEach.我可能最终调试这个(即如果我得到/花时间),最有可能通过锁定正确的部分,但是现在让我们说我想要阻止使用Parallel.ForEach或强制使用我的库以允许(或导致)只有一个线程,我该怎么办？

         try
         {
             ParallelOptions Options = new ParallelOptions();
             Options.CancellationToken = base.DownloadCancellation.Token;
             Parallel.ForEach(base.BlockingCollection1, Options, ActiveSeeder =>
                 {
                     //...
                 });
         }
         catch
         {
             if (base.DownloadCancellation.IsCancellationRequested)
                 return false;
         }

Parallel.Foreach/For 是否使用或不使用我放入 ParallelOptions 的 CancellationToken 调用 BlockingCollection1.Take 函数？

有机会知道吗？

你能解释为什么这段代码死机了吗？

int[] testlist = new int[ ] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
lock ( testlock ) {
    Parallel.ForEach( testlist, new ParallelOptions( ) { MaxDegreeOfParallelism = 90 }, ( int i ) => {
        Console.WriteLine( "hi there before " + i + " " + Monitor.IsEntered( testlock ) );
        lock ( testlock ) {
            Console.WriteLine( "hi there inner " + i + " " + Monitor.IsEntered( testlock ) );
        }
        Console.WriteLine( "hi there after …

我希望Timer运行5秒钟,同时向后端启动网络请求.我想在执行操作之前等待计时器和请求完成.

private async void doWork()
{
    await Task.Delay(5000);
    await doPostRequest();
    doSomethingElse();
}

像上面的代码一样,除了让这两个任务并行执行.我怎样才能做到这一点？

我正在尝试使用并行查询 mongo-db，Parallel.Foreach()但没有得到任何结果。但是当我尝试在常规 foreach 循环中运行相同的事情时，我能够执行预期的任务。

var exceptions = new ConcurrentQueue<Exception>();
var secondaryObjectsDictionaryCollection = new Dictionary<string, List<JObject>>();

// This works
foreach(var info in infos)
{
    try
    {
        name = await commonValidator.ValidateAsync(name);
        await commonValidator.ValidateIdAsync(name, id);
        var list = await helper.ListRelatedObjectsAsync(name, id, info, false);

        secondaryObjectsDictionaryCollection.Add(info.PrimaryId, secondaryObjectsList.ToList());
    }
    catch (Exception ex)
    {
        exceptions.Enqueue(ex);
    }
}

//This does not
Parallel.ForEach(infos, async info =>
{
    try
    {
        name = await commonValidator.ValidateAsync(name);
        await commonValidator.ValidateIdAsync(name, id);
        var list = await helper.ListRelatedObjectsAsync(name, id, info, false);

        secondaryObjectsDictionaryCollection.Add(info.PrimaryId, secondaryObjectsList.ToList());
    } …

我有一个程序,我想在其中模拟一个队列.为了加快速度(许多不同参数的)我认为我可以使用一个并行回路,然而队列对象(或者至少是对象中的对象),而它们都要么MGCC函数或队列内创建仍然共享宾语.我有没有忘记并行功能？

给出麻烦的对象是queue.MyHeap.

(如果需要更多信息,请询问,因为我遗漏了很多东西,使其更容易阅读,就像您在队列对象中看到的那样).

Parallel.ForEach(a, (numbers) =>
{                   
    MGcC(a);
});        

static public Tuple<Customer[,], List<Interval>[]> MGcC(int a)
{
    Queue queue = new Queue(a);
    return queue.Simulate(writeFile);
}

public class Queue
{
    Func<object, double> arrivalFunction;
    Func<object, double> servingFunction;
    double lambda;
    double v;
    object serviceObject;
    int minServers;
    bool decision;

    int idleServers;
    int activeServers;
    int amountInOrbit;
    protected minHeap myHeap;

    public Queue(double lambda, double v, object serviceObject, int servers, Func<object, double> arrivalFunction, Func<object, double> servingFunction, bool decision = false)
    {
        this.arrivalFunction = arrivalFunction;
        this.servingFunction = servingFunction;
        this.lambda …

我正在尝试同时下载多个文件。但是所有文件都在逐一下载。所以，首先下载这个文件@"http://download.geofabrik.de/europe/cyprus-latest.osm.pbf"，然后开始下载这个文件@"http://download.geofabrik.de/europe/finland-latest.osm.pbf",，下一个要下载的文件是@"http://download.geofabrik.de/europe/great-britain-latest.osm.pbf"等等。

但是我想同时下载。

因此，我根据此答案中的代码编写了以下代码：

static void Main(string[] args)
{
    Task.Run(async () =>
    {
        await DownloadFiles();
    }).GetAwaiter().GetResult();
}

public static async Task DownloadFiles()
{
    IList<string> urls = new List<string>
    {
        @"http://download.geofabrik.de/europe/cyprus-latest.osm.pbf",
        @"http://download.geofabrik.de/europe/finland-latest.osm.pbf",
        @"http://download.geofabrik.de/europe/great-britain-latest.osm.pbf",
        @"http://download.geofabrik.de/europe/belgium-latest.osm.pbf",
        @"http://download.geofabrik.de/europe/belgium-latest.osm.pbf"
    };

    foreach (var url in urls)
    {
        string fileName = url.Substring(url.LastIndexOf('/'));
        await DownloadFile(url, fileName);
    }            
}

public static async Task DownloadFile(string url, string fileName)
{
    string address = @"D:\Downloads";
    using (var client = new WebClient())
    {
        await …

我有一个在其中运行多个异步方法的方法。我必须遍历设备列表，并将设备传递给此方法。我注意到这需要很长时间才能完成，所以我正在考虑使用Parallel.ForEach它可以同时针对多个设备运行这个过程。

假设这是我的方法。

public async Task ProcessDevice(Device device) {
    var dev = await _deviceService.LookupDeviceIndbAsNoTracking(device);

    var result = await DoSomething(dev);
    await DoSomething2(dev);
}

然后 DoSomething2 也调用了一个异步方法。

public async Task DoSomething2(Device dev) {
    foreach(var obj in dev.Objects) {
        await DoSomething3(obj);
    }
}

随着时间的推移，设备列表不断变大，因此该列表增长得越多，程序完成ProcessDevice()对每个设备的运行所需的时间就越长。我想一次处理多个设备。所以我一直在研究使用Parallel.ForEach.

Parallel.ForEach(devices, async device => {
    try {
        await ProcessDevice(device);
    } catch (Exception ex) {
        throw ex;
    }
})

程序似乎在设备完全处理之前完成。我还尝试创建一个任务列表，然后为每个设备添加一个运行 ProcessDevice 的新任务到该列表，然后等待 Task.WhenAll(listOfTasks);

var listOfTasks = new List<Task>();
foreach(var device in devices) {
    var task …

我正在尝试将迭代 ArrayList 的普通 for 循环转换为并行循环。我遇到了parallelStream.forEach() 和parallelStream().collect(Collectors.toList()) 的概念。但不清楚如何转换。下面是普通 for 循环的示例代码：

public List<FinPlan> getFinPlanList() {
List<FinPlan> newList = new ArrayList<>();

// get list of string   
List<String> finPlanIdsList = finPlanSearchRequest.getFinPlanIds();

// iterate list and add element to new list
    for(String planId: finPlanIdsList) {
    
        newList.add(retrieveFinPlan(planId));
    }
    
    return newList;
}

我尝试使用以下代码，但它给了我错误“newList 应该是最终的”。

    finPlanIdsList.parallelStream().forEach( planId -> {
            newList.add(retrieveFinPlan(planId));
    });

那么，主要问题是如何将上面的普通循环转换为parallelStream.forEach()和parallelStream().collect()？

我对LINQ“ order by”有一个非常奇怪的问题，并且 Parallel.Foreach

具体来说，此代码有效：

IList<IEntitaAssociabile> result = new List<IEntitaAssociabile>();

foreach(PraticheAperteNonAssegnate pratica in praticheAperteNonAssegnate)
{
    result.Add(new EntitaAssociabile
    {
          Id = pratica.ID_Prat,
          TipologiaEntita = TipologiaEntita.Pratica,
          DataApertura = pratica.DataAper.Value,
          TipologiaPratica = pratica.Cod_TpPrat,
          NomeCliente = pratica.Nominativo,
          NumeroPraticheDaAssociare = null,
          TipologiaEntitaPadre = GetEntitaPadre(pratica, praticheLotti, praticheSottolotti),
          IdEntitaPadre = GetIdEntitaPadre(pratica, praticheLotti, praticheSottolotti)
    });
}

return result.OrderBy(x => x.Id).ToList();

如果我只是用以下命令更改foreach语句Parallel.Foreach：

 IList<IEntitaAssociabile> result = new List<IEntitaAssociabile>();

 Parallel.ForEach(praticheAperteNonAssegnate, (pratica) =>
 {
      result.Add(new EntitaAssociabile
      {
           Id = pratica.ID_Prat,
           TipologiaEntita = TipologiaEntita.Pratica,
           DataApertura = pratica.DataAper.Value,
           TipologiaPratica = …