内存中的简单消息队列

Question

我们现有的域事件实现限制（通过阻止）一次发布到一个线程，以避免对处理程序的重入调用：

public interface IDomainEvent {}  // Marker interface
public class Dispatcher : IDisposable
{
    private readonly SemaphoreSlim semaphore = new SemaphoreSlim(1, 1);
    // Subscribe code...
    public void Publish(IDomainEvent domainEvent)
    {
        semaphore.Wait();
        try
        {
            // Get event subscriber(s) from concurrent dictionary...
            foreach (Action<IDomainEvent> subscriber in eventSubscribers)
            {
                subscriber(domainEvent);
            }
        }
        finally
        {
            semaphore.Release();
        }
    }
    // Dispose pattern...
}

如果处理程序发布了一个事件，这将导致死锁。

如何重写它以序列化对Publish的调用？换句话说，如果订阅处理程序A发布事件B，我将得到：

1. 已呼叫处理程序A
2. 已呼叫处理程序B

同时在多线程环境中保留对处理程序没有可重入调用的条件。

我不想更改公共方法签名；例如，应用程序中没有位置可以调用方法来发布队列。

Answer 1

您必须使发布异步才能实现这一点。简单的实现方式如下：

public class Dispatcher : IDisposable {
    private readonly BlockingCollection<IDomainEvent> _queue = new BlockingCollection<IDomainEvent>(new ConcurrentQueue<IDomainEvent>());
    private readonly CancellationTokenSource _cts = new CancellationTokenSource();
    public Dispatcher() {
        new Thread(Consume) {
            IsBackground = true
        }.Start();
    }
    private List<Action<IDomainEvent>> _subscribers = new List<Action<IDomainEvent>>();
    public void AddSubscriber(Action<IDomainEvent> sub) {
        _subscribers.Add(sub);
    }
    private void Consume() {            
        try {
            foreach (var @event in _queue.GetConsumingEnumerable(_cts.Token)) {
                try {
                    foreach (Action<IDomainEvent> subscriber in _subscribers) {
                        subscriber(@event);
                    }
                }
                catch (Exception ex) {
                    // log, handle                        
                }
            }
        }
        catch (OperationCanceledException) {
            // expected
        }
    }
    public void Publish(IDomainEvent domainEvent) {
        _queue.Add(domainEvent);
    }
    public void Dispose() {
        _cts.Cancel();
    }
}

Answer 2

我们想出了一种同步实现的方法。

public class Dispatcher : IDisposable
{
    private readonly ConcurrentQueue<IDomainEvent> queue = new ConcurrentQueue<IDomainEvent>();
    private readonly SemaphoreSlim semaphore = new SemaphoreSlim(1, 1);
    // Subscribe code...
    public void Publish(IDomainEvent domainEvent)
    {
        queue.Enqueue(domainEvent);
        if (IsPublishing)
        {
            return;
        }
        PublishQueue();
    }
    private void PublishQueue()
    {
        IDomainEvent domainEvent;
        while (queue.TryDequeue(out domainEvent))
        {
            InternalPublish(domainEvent);
        }
    }
    private void InternalPublish(IDomainEvent domainEvent)
    {
        semaphore.Wait();
        try
        {
            // Get event subscriber(s) from concurrent dictionary...
            foreach (Action<IDomainEvent> subscriber in eventSubscribers)
            {
                subscriber(domainEvent);
            }
        }
        finally
        {
            semaphore.Release();
        }
        // Necessary, as calls to Publish during publishing could have queued events and returned.
        PublishQueue();
    }
    private bool IsPublishing
    {
        get { return semaphore.CurrentCount < 1; }
    }
    // Dispose pattern for semaphore...
}

}

Answer 3

这个接口是做不到的。您可以异步处理事件订阅以消除死锁，同时仍然串行运行它们，但不能保证所描述的顺序。另一个对Publish的调用可能会在事件A的处理程序正在运行但在发布事件B之前将某个东西（事件C）排入队列。然后，事件B最终会排在队列中的事件C之后。

只要处理程序A在队列中获取项目时与其他客户端处于平等地位，它要么必须像其他人一样等待（死锁），要么必须公平地进行（先到先得）。您的界面不允许将两者区别对待。

这并不是说你不能在逻辑中进行一些恶作剧来试图区分它们（例如，基于线程id或其他可识别的东西），但如果你不控制订阅者代码，那么这些方面的任何东西都是不可靠的。