标签: parallel.foreach

如何在Parallel.Foreach中定义计数器并停止以特定数量消耗？
我问了这个问题,因为Parallel.ForEach中的那个计数器在常规操作中不起作用.
请看这个小例子:

static void Main(string[] args)
{
    int Count_Step = -1;
    string[] lines = new string[] 
    { 
        "0",
        "1",
        "2",
        "3",
        "4",
        "5",
        "6",
        "7",
        "8",
        "9",
        "10",
        "11",
        "12",
        "13",
        "14",
        "15",
        "16",
        "17",
        "18",
        "19"
    };

    List<string> list_lines = new List<string>(lines);

    ParallelOptions parallelOptions = new ParallelOptions();
    parallelOptions.MaxDegreeOfParallelism = 3;

    Parallel.ForEach(list_lines, parallelOptions, (line, state, index) =>
    {
        if (Count_Step == 10)
            state.Stop();
        Count_Step++;
        Console.WriteLine(index + " : " + line + " : " + Count_Step); …

我正在实现一个Parallel.ForEach循环来完成一些工作,但是由于未处理的异常而导致了问题,而我认为我处理了取消.

为了尝试解决这个问题,我在winform中进行了一个简单的测试设置.它有一个开始按钮,一个取消按钮和一个输出标签.

代码:

public partial class Form1 : Form
{
    CancellationTokenSource cts = new CancellationTokenSource();

    public Form1()
    {
        InitializeComponent();
    }

    private async void button1_Click(object sender, EventArgs e)
    {
        output.Text = "Running";

        try
        {
            var runTask = Task<string>.Factory.StartNew(() => Run());
            await runTask;
            this.output.Text = runTask.Result;
        }
        catch(Exception ex)
        {
            throw ex;
        }
    }

    private string Run()
    {
        int useThreads = Environment.ProcessorCount - 2 < 1 ? 1 : Environment.ProcessorCount - 2;

        ParallelOptions options = new ParallelOptions() { MaxDegreeOfParallelism = useThreads, CancellationToken …

我需要从网站上抓取数据。我有 1,000 多个链接需要访问，以前我将每个线程划分为 10 个链接，然后开始 100 个线程，每个线程拉取 10 个。经过几个测试用例后，100 个线程是最好的计数，以最大限度地减少它检索内容的时间所有的链接。

我意识到 .NET 4.0 为开箱即用的多线程提供了更好的支持，但这是根据您拥有的内核数量来完成的，在我的情况下，这不会产生足够的线程。我想我要问的是：优化 1,000 个链接拉动的最佳方法是什么。我应该使用.ForEach并让Parallel扩展控制产生的线程数量，还是找到一种方法来告诉它启动和划分工作的线程数？

我以前没有工作过，Parallel所以也许我的方法可能是错误的。

我已经使用Parallel.ForEach功能来同时处理多个客户端,如下所示:

Clients objClient = new Clients();
List<Clients> objClientList = Clients.GetClientList();

Parallel.ForEach<Clients>(objClientList, list =>
{
    SendFilesToClient(list);
});

但是现在,我没有创建Client类,而是决定动态创建客户端对象,如下所示:

var xDoc = XDocument.Load(new StreamReader(xmlPath + @"\client.xml"));
dynamic root = new ExpandoObject();

XmlToDynamic.Parse(root, xDoc.Elements().First());

dynamic clients = new List<dynamic>();

for (int i = 0; i < root.clients.client.Count; i++)
{
    clients.Add(new ExpandoObject());
    clients[i].Id = root.clients.client[i].id;
    clients[i].Name = root.clients.client[i].name;
    List<string> list = new List<string>();
    for (int j = 0; j < root.clients.client[i].emails.email.Count; j++)
    {
        list.Add(root.clients.client[i].emails.email[j].ToString());
    }
    clients[i].Email = string.Join(",", list);
}

现在我不使用Client类并动态生成对象,如何使用Parallel.ForEach功能同时处理动态对象中的多个客户端,就像我以前使用类对象一样？

希望我清楚自己.

此外,如果您能告诉我我的方法是否有问题或者告诉我更好的方法,我将不胜感激.

我有几种不同的方法可以将整个目录上传到我的应用程序中的Amazon S3,具体取决于所选的选项.目前,其中一个选项将并行执行多个目录的上载.我不确定这是不是一个好主意,因为在某些情况下它会加快上传速度,而其他情况则会减慢它的速度.加速似乎是在有一堆小目录时,但如果批处理中有大型目录,则速度会变慢.我正在使用下面看到的并行ForEach循环并使用AWS API的TransferUtility.UploadDirectoryAsync()方法:

Parallel.ForEach(dirs,myParallelOptions, 
                   async dir => { await MyUploadMethodAsync(dir) };

TransferUtility.UploadDirectoryAsync()方法在哪里MyUploadMethodAsync().这些TransferUtility上传方法都执行单个文件的并行上传(如果大小足够大),所以执行目录的并行上传可能是过度的.显然,我们仍然限制可用的带宽量,因此这可能是一种浪费,我应该只使用该UploadDirectoryAsync()方法的常规foreach循环.任何人都可以提供一些见解,如果这是并行化的坏情况？

我有以下代码:

HttpContext httpContext = HttpContext.Current;
RequestContext currentContext = RequestContextManager.CurrentContext;
ILifetimeScope currentSessionScope = PlatformContext.LifeTimeScope;

ConcurrentQueue<Exception> exceptions = new ConcurrentQueue<Exception>();
ConcurrentBag<ParallelCalculateObj> forEachResult = new ConcurrentBag<ParallelCalculateObj>();
ConcurrentBag<ParallelCalculateObj> testForEachPassResult = new ConcurrentBag<ParallelCalculateObj>();

ParallelLoopResult loopResult = Parallel.ForEach(applications, () =>
{
    HttpContext.Current = httpContext;
    RequestContextManager.SetCustomCurrentContext(currentContext);
    PlatformContext.LifeTimeScope = currentSessionScope;
    return new ParallelCalculateObj();
}, (application, pls, localObj) =>
{
    try
    {
        // some code
    }
    catch (Exception e)
    {
        exceptions.Enqueue(e);
    }
    testForEachPassResult.Add(localObj);
    return localObj;
}, forEachResult.Add);

哪里applications.Count = 3.上面的代码执行后,我得到了forEachResult.Count = 2和testForEachPassResult.Count = 3 …

我有100条并行化记录,从1到100,现在我可以方便地使用Parallel.For在Parallel中执行它们如下,这将基于计算资源工作

 Parallel.For(0, limit, i =>
    {
        DoWork(i);
    });

但是有一些限制,每个线程需要使用相同的数据实体,并且数量有限的数据实体说10,它们是通过相互克隆并将它们保存在像Dictionary或List这样的结构中而先进创建的.现在我可以使用以下代码限制并行化的数量:

 Parallel.For(0, limit, new ParallelOptions { MaxDegreeOfParallelism = 10 }, i =>
    {
        DoWork(i);
    });

但问题是如何为每个传入线程分配一个唯一的数据实体,这样任何其他当前线程都不会使用Data实体,因为线程和数据实体的数量是相同的,所以饥饿不是问题.我可以想到,我为每个数据实体创建一个布尔值,指定它是否正在使用,因此我们遍历字典或列表以查找下一个可用数据实体并锁定整个分配过程,以便一个线程在给定时间被分配了一个数据实体,但在我看来这个问题将有更优雅的解决方案,我的版本只是一个解决方法,而不是真正的修复.我的逻辑是:

Parallel.For(0, limit, new ParallelOptions { MaxDegreeOfParallelism = 10 }, i =>
        {
            lock(All_Threads_Common_Object)
            {
              Check for available data entity using boolean
              Assign the Data entity
            }
            DoWork(i);

            Reset the Boolean value for another thread to use it
        });

如果问题需要进一步澄清,请告诉我

以下代码中是否存在可能的竞争条件？

public void Process(List<SomeObject> list)
{
    SomeDataOutput objData=null;
    ConcurrentBag<SomeDataOutput> cbOutput = new ConcurrentBag<SomeDataOutput>();
    ParallelOptions po = new ParallelOptions(){MaxDegreeOfParallelism=4};
    Parallel.ForEach(list, po, (objInput) =>
    {
        objData = GetOutputData(objInput);//THIS LINE IS THE ONE I AM UNSURE OF. CAN objData GET OVERWRITTEN BY MULTIPLE PARALLEL THREADS?
        cbOutput.Add(objData);
    });
}  

今天我遇到了一个我无法解释的奇怪现象.在gridview中有一个包含多个行的网页,需要将其逐个保存到数据库和XML文件中.我最终使用了一个Parallel.ForEach,因为行之间没有关系,所以它们可以独立执行.代码基本上是这样的:

        Parallel.ForEach(gvWithData.Rows.Cast<GridViewRow>(), row =>
        {
            if (row.RowType == DataControlRowType.DataRow)
            {
                    // do some logic and stuff...
                    var type = new Object { ... };

                    // save to the database
                    type.Save();

                    // retrieve the saved item from the database again
                    //  since we need some autoincrement values from the db
                    var typeAfterSave = TypeManager.GetFromDb();

                    // create a custom XML from the object
                    XmlManager.CreateXml(typeAfterSave);
                }
            }

为什么这个代码在我Parallel.ForEach用一个好老的替换时会有任何不同之处,foreach而我什么都不做呢？

不同之处在于,在第一种情况下创建XML的文化与第二种情况不同,我没有丝毫的线索.

有什么建议？

我有一个有序列表,如[1, 2, 3, 4, 5, 6, 7, 8, 9, 10].我将它传递给一份Parallel.ForEach声明.我可以以某种方式实现以下桶的执行顺序,例如:处理前3个项目[1, 2, 3],其中桶本身的排序不是强制性的,[2, 1, 3]例如可以.然后处理接下来的3个项目[4, 5, 6]等？