在多个线程中访问 List<T> 元素并报告进度

【问题标题】：Access List<T> elements in multiple threads and report progress在多个线程中访问 List<T> 元素并报告进度
【发布时间】：2013-01-11 19:59:27
【问题描述】：

在我的应用程序中，我有来自我的数据库的人员列表。 对于每个人，我必须调用 5 个（目前）服务来搜索一些信息。 如果服务返回信息，我会将其添加到该人（特定人的订单列表）
因为服务独立工作，我想我可以尝试并行运行它们。 我已经这样创建了我的代码：

using System;
using System.Collections.Generic;
using System.Threading;

namespace Testy
{
    internal class Program
    {
        internal class Person
        {
            public int Id { get; set; }
            public string Name { get; set; }
            public List<string> Orders { get; private set; }

            public Person()
            {
                // thanks for tip @juharr
                Orders = new List<string>();
            }

            public void AddOrder(string order)
            {
                lock (Orders) //access across threads
                {
                    Orders.Add(order);
                }
            }
        }

        internal class Service
        {
            public int Id { get; private set; }

            public Service(int id)
            {
                Id = id;
            }

            //I get error when I use IList instead of List
            public void Search(ref List<Person> list) 
            {
                foreach (Person p in list)
                {
                    lock (p) //should I lock Person here? and like this???
                    {
                        Search(p);
                    }
                }
            }
            private void Search(Person p)
            {
                Thread.Sleep(50);
                p.AddOrder(string.Format("test order from {0,2}",
                                      Thread.CurrentThread.ManagedThreadId));
                Thread.Sleep(100);
            }
        }

        private static void Main()
        {
            //here I load my services from external dll's
            var services = new List<Service>();
            for (int i = 1; i <= 5; i++)
            {
                services.Add(new Service(i));
            }

            //sample data load from db    
            var persons = new List<Person>();

            for (int i = 1; i <= 10; i++)
            {
                persons.Add(
                    new Person {Id = i, 
                    Name = string.Format("Test {0}", i)});
            }

            Console.WriteLine("Number of services: {0}", services.Count);
            Console.WriteLine("Number of persons: {0}", persons.Count);

            ManualResetEvent resetEvent = new ManualResetEvent(false);
            int toProcess = services.Count;

            foreach (Service service in services)
            {
                new Thread(() =>
                    {
                        service.Search(ref persons);
                        if (Interlocked.Decrement(ref toProcess) == 0)
                            resetEvent.Set();
                    }
                    ).Start();
            }

            // Wait for workers.
            resetEvent.WaitOne();

            foreach (Person p in persons)
            {
                Console.WriteLine("{0,2} Person name: {1}",p.Id,p.Name);
                if (null != p.Orders)
                {
                    Console.WriteLine("    Orders:");
                    foreach (string order in p.Orders)
                    {
                        Console.WriteLine("    Order: {0}", order);
                    }
                }
                else
                {
                    Console.WriteLine("    No orders!");
                }
            }
            Console.ReadLine();
        }
    }
}

我的代码有 2 个问题：

  1. 当我运行我的应用程序时，我应该得到 10 人的列表，每个人有 5 个订单，但有时（3-5 次运行的）第一人我只得到 4 个订单。 如何防止此类行为？
    解决了！感谢@juharr
  2. 如何从我的线程报告进度？我想得到的是我的程序类中的一个函数，每次从服务中添加订单时都会调用它——我需要它来显示每个报告的某种进展。
    我正在尝试此处描述的解决方案：https://stackoverflow.com/a/3874184/965722，但我想知道是否有更简单的方法。理想情况下，我想将委托添加到 Service 类并将所有线程代码放在那里。
    我应该如何将事件和委托添加到 Service 类以及如何在 Main 方法中订阅它？

我使用的是 .NET 3.5

我已添加此代码以获取进度报告：

using System;
using System.Collections.Generic;
using System.Threading;

namespace Testy
{
    internal class Program
    {
        public class ServiceEventArgs : EventArgs
        {
            public ServiceEventArgs(int sId, int progress)
            {
                SId = sId;
                Progress = progress;
            }

            public int SId { get; private set; }
            public int Progress { get; private set; }
        }

        internal class Person
        {
            private static readonly object ordersLock = new object();

            public int Id { get; set; }
            public string Name { get; set; }
            public List<string> Orders { get; private set; }

            public Person()
            {
                Orders = new List<string>();
            }

            public void AddOrder(string order)
            {
                lock (ordersLock) //access across threads
                {
                    Orders.Add(order);
                }
            }
        }

        internal class Service
        {
            public event EventHandler<ServiceEventArgs> ReportProgress;

            public int Id { get; private set; }
            public string Name { get; private set; }

            private int counter;

            public Service(int id, string name)
            {
                Id = id;
                Name = name;
            }

            public void Search(List<Person> list) //I get error when I use IList instead of List
            {
                counter = 0;
                foreach (Person p in list)
                {
                    counter++;
                    Search(p);
                    Thread.Sleep(3000);
                }
            }

            private void Search(Person p)
            {
                p.AddOrder(string.Format("Order from {0,2}", Thread.CurrentThread.ManagedThreadId));

                EventHandler<ServiceEventArgs> handler = ReportProgress;
                if (handler != null)
                {
                    var e = new ServiceEventArgs(Id, counter);
                    handler(this, e);
                }
            }
        }

        private static void Main()
        {
            const int count = 5;
            var services = new List<Service>();
            for (int i = 1; i <= count; i++)
            {
                services.Add(new Service(i, "Service " + i));
            }

            var persons = new List<Person>();

            for (int i = 1; i <= 10; i++)
            {
                persons.Add(new Person {Id = i, Name = string.Format("Test {0}", i)});
            }

            Console.WriteLine("Number of services: {0}", services.Count);
            Console.WriteLine("Number of persons: {0}", persons.Count);
            Console.WriteLine("Press ENTER to start...");
            Console.ReadLine();

            ManualResetEvent resetEvent = new ManualResetEvent(false);
            int toProcess = services.Count;

            foreach (Service service in services)
            {
                new Thread(() =>
                    {
                        service.ReportProgress += service_ReportProgress;
                        service.Search(persons);
                        if (Interlocked.Decrement(ref toProcess) == 0)
                            resetEvent.Set();
                    }
                    ).Start();
            }

            // Wait for workers.
            resetEvent.WaitOne();

            foreach (Person p in persons)
            {
                if (p.Orders.Count != count)
                    Console.WriteLine("{0,2} Person name: {1}, orders: {2}", p.Id, p.Name, p.Orders.Count);
            }
            Console.WriteLine("END :)");
            Console.ReadLine();
        }

        private static void service_ReportProgress(object sender, ServiceEventArgs e)
        {
            Console.CursorLeft = 0;
            Console.CursorTop = e.SId;
            Console.WriteLine("Id: {0,2}, Name: {1,2} - Progress: {2,2}", e.SId, ((Service) sender).Name, e.Progress);
        }
    }
}

我为服务类添加了自定义 EventArgs，事件。 在此配置中，我应该运行 5 个服务，但其中只有 3 个报告进度。
我想如果我有 5 个服务，我应该有 5 个事件（5 行显示进度）。
这可能是因为线程，但我不知道如何解决这个问题。

示例输出现在如下所示：

Number of services: 5
Number of persons: 10
Press ENTER to start...
Id:  3, Name: Service 3 - Progress: 10
Id:  4, Name: Service 4 - Progress: 10
Id:  5, Name: Service 5 - Progress: 19
END :)

应该是这样的：

Number of services: 5
Number of persons: 10
Press ENTER to start...
Id:  1, Name: Service 1 - Progress: 10
Id:  2, Name: Service 2 - Progress: 10
Id:  3, Name: Service 3 - Progress: 10
Id:  4, Name: Service 4 - Progress: 10
Id:  5, Name: Service 5 - Progress: 10
END :)

上次编辑
我已将所有线程创建移至单独的类 ServiceManager 现在我的代码如下所示：

using System;
using System.Collections.Generic;
using System.Threading;

namespace Testy
{
    internal class Program
    {
        public class ServiceEventArgs : EventArgs
        {
            public ServiceEventArgs(int sId, int progress)
            {
                SId = sId;
                Progress = progress;
            }

            public int SId { get; private set; } // service id
            public int Progress { get; private set; }
        }

        internal class Person
        {
            private static readonly object ordersLock = new object();

            public int Id { get; set; }
            public string Name { get; set; }
            public List<string> Orders { get; private set; }

            public Person()
            {
                Orders = new List<string>();
            }

            public void AddOrder(string order)
            {
                lock (ordersLock) //access across threads
                {
                    Orders.Add(order);
                }
            }
        }

        internal class Service
        {
            public event EventHandler<ServiceEventArgs> ReportProgress;

            public int Id { get; private set; }
            public string Name { get; private set; }

            public Service(int id, string name)
            {
                Id = id;
                Name = name;
            }

            public void Search(List<Person> list)
            {
                int counter = 0;
                foreach (Person p in list)
                {
                    counter++;
                    Search(p);
                    var e = new ServiceEventArgs(Id, counter);
                    OnReportProgress(e);
                }
            }

            private void Search(Person p)
            {
                p.AddOrder(string.Format("Order from {0,2}", Thread.CurrentThread.ManagedThreadId));
                Thread.Sleep(50*Id);
            }

            protected virtual void OnReportProgress(ServiceEventArgs e)
            {
                var handler = ReportProgress;
                if (handler != null)
                {
                    handler(this, e);
                }
            }
        }

        internal static class ServiceManager
        {
            private static IList<Service> _services;

            public static IList<Service> Services
            {
                get
                {
                    if (null == _services)
                        Reload();
                    return _services;
                }
            }

            public static void RunAll(List<Person> persons)
            {
                ManualResetEvent resetEvent = new ManualResetEvent(false);
                int toProcess = _services.Count;

                foreach (Service service in _services)
                {
                    var local = service;
                    local.ReportProgress += ServiceReportProgress;
                    new Thread(() =>
                        {
                            local.Search(persons);
                            if (Interlocked.Decrement(ref toProcess) == 0)
                                resetEvent.Set();
                        }
                        ).Start();
                }
                // Wait for workers.
                resetEvent.WaitOne();
            }

            private static readonly object consoleLock = new object();

            private static void ServiceReportProgress(object sender, ServiceEventArgs e)
            {
                lock (consoleLock)
                {
                    Console.CursorTop = 1 + (e.SId - 1)*2;
                    int progress = (100*e.Progress)/100;
                    RenderConsoleProgress(progress, '■', ConsoleColor.Cyan, String.Format("{0} - {1,3}%", ((Service) sender).Name, progress));
                }
            }

            private static void ConsoleMessage(string message)
            {
                Console.CursorLeft = 0;
                int maxCharacterWidth = Console.WindowWidth - 1;
                if (message.Length > maxCharacterWidth)
                {
                    message = message.Substring(0, maxCharacterWidth - 3) + "...";
                }
                message = message + new string(' ', maxCharacterWidth - message.Length);
                Console.Write(message);
            }

            private static void RenderConsoleProgress(int percentage, char progressBarCharacter,
                                                      ConsoleColor color, string message)
            {
                ConsoleColor originalColor = Console.ForegroundColor;
                Console.ForegroundColor = color;
                Console.CursorLeft = 0;
                int width = Console.WindowWidth - 1;
                var newWidth = (int) ((width*percentage)/100d);
                string progBar = new string(progressBarCharacter, newWidth) + new string(' ', width - newWidth);
                Console.Write(progBar);
                if (!String.IsNullOrEmpty(message))
                {
                    Console.CursorTop++;
                    ConsoleMessage(message);
                    Console.CursorTop--;
                }
                Console.ForegroundColor = originalColor;
            }

            private static void Reload()
            {
                if (null == _services)
                    _services = new List<Service>();
                else
                    _services.Clear();

                for (int i = 1; i <= 5; i++)
                {
                    _services.Add(new Service(i, "Service " + i));
                }
            }
        }

        private static void Main()
        {
            var services = ServiceManager.Services;
            int count = services.Count;

            var persons = new List<Person>();

            for (int i = 1; i <= 100; i++)
            {
                persons.Add(new Person {Id = i, Name = string.Format("Test {0}", i)});
            }

            Console.WriteLine("Services: {0}, Persons: {1}", services.Count, persons.Count);
            Console.WriteLine("Press ENTER to start...");
            Console.ReadLine();
            Console.Clear();
            Console.CursorVisible = false;

            ServiceManager.RunAll(persons);

            foreach (Person p in persons)
            {
                if (p.Orders.Count != count)
                    Console.WriteLine("{0,2} Person name: {1}, orders: {2}", p.Id, p.Name, p.Orders.Count);
            }
            Console.CursorTop = 12;
            Console.CursorLeft = 0;
            Console.WriteLine("END :)");
            Console.CursorVisible = true;
            Console.ReadLine();
        }
    }
}

【问题讨论】：

  • 您的空检查和订单列表的创建应该在锁内，或者更好的是在 Person 构造函数内。
  • @juharr - 感谢您的提示 :) 我会立即修改并检查我的代码！

标签： c# multithreading .net-3.5


【解决方案1】：

基本上，您在创建订单时会遇到竞争条件。想象一下以下两个线程的执行。

线程 1 检查 Orders 是否为空。
线程 2 检查 Orders 是否为空。
线程 1 将 Orders 设置为一个新列表。
线程 1 获得锁。
线程 1 添加到订单列表。
线程 2 将 Order 设置为一个新列表。 （你只是丢失了线程 1 添加的内容）

您需要在锁中包含订单的创建。

public void AddOrder(string order)
{
    lock (Orders) //access across threads
    {
        if (null == Orders)
            Orders = new List<string>();
        Orders.Add(order);
    }
}

或者你真的应该在 Person 构造函数中创建 Order 列表

public Person()
{
    Orders = new List<Order>();
}

另外，你真的应该为锁定创建一个单独的对象。

private object ordersLock = new object();


public void AddOrder(string order)
{
    lock (ordersLock) //access across threads
    {
        Orders.Add(order);
    }
}

编辑：

在您创建线程的 foreach 中，您需要创建服务的本地副本以在 lambda 表达式中使用。这是因为 foreach 将更新服务变量，而线程最终可能会捕获错误的变量。所以像这样。

foreach (Service service in services)
{
    Service local = service;
    local.ReportProgress += service_ReportProgress;
    new Thread(() =>
        {
            local.Search(persons);
            if (Interlocked.Decrement(ref toProcess) == 0)
                resetEvent.Set();
        }
    ).Start();
}

请注意，订阅不需要在线程内。

或者，您可以将线程的创建移动到您的 Service 类的 Search 方法中。

此外，您可能希望在 Service 类中创建一个 OnReportProgress 方法，如下所示：

protected virtual void OnReportProgress(ServiceEventArgs e)
{
    EventHandler<ServiceEventArgs> handler = ReportProgress;
    if (handler != null)
    {
        handler(this, e);
    }
}

然后在您的 Search 方法中调用它。就我个人而言，我会在公共 Search 方法中调用它，并将计数器设为局部变量，以允许在另一个列表中重用 Service 对象。

最后，在写入控制台时，您需要在事件处理程序中添加一个额外的锁，以确保一个线程在另一个线程写入输出之前不会更改光标位置。

private static object consoleLock = new object();

private static void service_ReportProgress(object sender, ServiceEventArgs e)
{
    lock (consoleLock)
    {
        Console.CursorLeft = 0;
        Console.CursorTop = e.SId;
        Console.WriteLine("Id: {0}, Name: {1} - Progress: {2}", e.SId, ((Service)sender).Name, e.Progress);
    }
}

您可能还想在以下位置使用Console.Clear()：

...
Console.WriteLine("Number of services: {0}", services.Count);
Console.WriteLine("Number of persons: {0}", persons.Count);
Console.WriteLine("Press ENTER to start...");
Console.Clear();
Console.ReadLine();
...

您需要在写出结束语句之前更新光标位置。

Console.CursorTop = 6;
Console.WriteLine("END :)");

【讨论】：

  • 感谢您的提示 :) 我已经更改了 person 课程，但我想知道那个储物柜。如果它们位于单独的 dll 中，我将如何在我的服务中使用它？我应该在主应用程序中创建一个对象并使用 ref 将其传递给服务吗？
  • 锁定对象将在您的 Person 类中。所以每个Person 都会有一个锁定对象private 用于您的AddOrder 方法。
  • 所以我不需要在Service 类中的Search 方法中使用另一个锁？请在我的问题中查看我编辑的 Service 类代码
  • 不，您不需要 Search 方法中的锁，因为您没有修改 Person 列表。但是，那里的锁可以解决问题，因为它限制一个Service 一次更新一个Person。通常，您希望锁定在最低级别。是的，锁定对象可以是readonly。
  • 您可以在 Service 类中创建一个事件，然后在开始搜索之前从 Main 订阅它。然后在 Search 方法中使用自定义事件参数类中所需的信息调用事件。这是一篇关于编写事件的好文章msdn.microsoft.com/en-us/library/w369ty8x(v=vs.80).aspx
【解决方案2】：

这可能无法完全回答您的问题（但我认为您可能有竞争条件），当您开始处理线程时，您需要在从不同线程更新对象时实现适当的同步。您应该确保在任何给定时间只有一个线程能够更新 person 类的实例。 p.AddOrder( 应该有互斥锁，确保只有一个线程在更新 Person 对象。

【讨论】：

  • 我不能用lock代替mutex吗？
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2021-11-21
  • 2012-11-09
  • 2019-04-30
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode