多事件和多线程答案

【问题标题】：Multiple events and multi threading多事件和多线程
【发布时间】：2015-12-30 09:37:47
【问题描述】：

我有一个引发多个事件的服务，其中一些可以同时引发。我需要处理这些事件并根据事件参数运行可能需要长时间运行的方法。
我所做的是创建一个BlockingCollection<T>，它将事件存储在一个Task 中，它将继续一次处理一个事件，直到收到停止使用CancellationTokenSource 的信号。
我担心的是我没有很好地处理同步。
这是处理所有事情的类（它被用作WPF ViewModel）：

public class EventsTest
{
    //private fields
    private BlockingCollection<IoEventArgs> _queue;
    private CancellationTokenSource _tokenSource;
    private IoService _ioService;
    private Task _workerTask;
    private static EventWaitHandle _eventWaiter;

    public EventsTest()
    {
        _queue = new BlockingCollection<IoEventArgs>();
        _tokenSource = new CancellationTokenSource();
        _eventWaiter = new EventWaitHandle(false, EventResetMode.AutoReset);

        //this is the object that raises multiple events
        _ioService = new IoService();
        _ioService.IoEvent += _ioService_IoEvent;

        //Start Listening
        var t = Task.Factory.StartNew(StartListening, _tokenSource, TaskCreationOptions.LongRunning);
    }

    //IO events listener
    private void _ioService_IoEvent(string desc, int portNum)
    {
        //add events to a blocking collection
        _queue.Add(new IoEventArgs() { Description = desc, PortNum = portNum });
    }

    private void StartListening(object dummy)
    {
        //process the events one at a time
        while (!_tokenSource.IsCancellationRequested)
        {
            var eve = _queue.Take();
            switch (eve.PortNum)
            {
                case 0:
                    LongRunningMethod(eve.Description);
                    break;
                case 1:
                    //invoke a long running method
                    break;
                default:
                    break;
            }
        }
    }

    //sample long running method
    private void LongRunningMethod(string data)
    {
        _eventWaiter.WaitOne(10000);
    }
}

我怎样才能使这个过程在线程安全方面更加健壮？
在每个方法实现周围添加lock 会提高过程的安全性吗？

【问题讨论】：

标签： c# wpf multithreading events .net-4.6

【解决方案1】：

您的.Take() 不会被取消，因此您可能会一直在那里等待。

您可以将令牌传递给：

var eve = _queue.Take(_tokenSource);

但是你必须处理异常。

更好的方法是 TryTake(out eve, 1000, _tokenSource) 并使用返回的布尔值进行引导。

或者忘记 CancellationToken 并使用 AddingComplete()

【讨论】：

不允许传递_tokenSource：cannot convert from 'System.Threading.CancellationTokenSource' to 'System.Threading.CancellationToken'。我需要为此创建一个单独的CancellationToken 吗？
是的，Source 有一个 .Token 属性。

【解决方案2】：

这听起来像是微软的反应式框架更适合的情况。

您的代码如下所示：

public class EventsTest
{
    private IDisposable _subscription;

    public EventsTest()
    {
        IoService ioService = new IoService();

        _subscription =
            Observable
                .FromEvent<IoEvent, IoEventArgs>(
                    a => ioService.IoEvent += a, a => ioService.IoEvent -= a)
                .Subscribe(eve =>
                {
                    switch (eve.PortNum)
                    {
                        case 0:
                            LongRunningMethod(eve.Description);
                            break;
                        case 1:
                            //invoke a long running method
                            break;
                        default:
                            break;
                    }
                });
    }

    private void LongRunningMethod(string data)
    {
    }
}

这应该自动确保多个事件排队并且永远不会重叠。如果有问题，只需在.Subscribe(...) 之前拨打.Synchronize() 电话，它就会完美运行。

当您想取消活动时，只需致电_subscription.Dispose()，一切都会为您清理干净。

NuGet "Rx-Main" 获取您需要的位。

【讨论】：