CQRS架构，C端的职责是处理从上层发送过来的command。对于单台机器来说，我们如何尽快的处理command呢？本文想通过不断提问和回答的方式，把我的思考写出来。

首先，我们最容易想到的是使用多线程。那当我们要处理一个command时，能直接丢到线程池中，直接交给线程池去调度吗？不行。因为假如多个command修改同一个聚合根时，会导致db的并发冲突，从而会导致command的不断重试，大大降低了command的处理速度。

那该怎么解决呢？既然直接多线程处理会有并发冲突的问题，那就对command进行路由。那根据什么进行路由呢？考虑到大部分command都会指定要创建或修改哪个聚合根，也就是说会指定聚合根ID。所以，我们就很容易想到，可以根据聚合根ID进行路由。我们可以根据聚合根ID的hashcode得到一个long；然后我们系统启动时，创建N个ConcurrentQueue。然后路由就很简单了，我们可以对聚合根ID的hashcode对N取模，然后得到余数，然后该余数就是对应的queue的索引，然后我们把该command放入对应的queue即可。然后每个queue的消费者，有一个单线程在不断的按次序逐个消费当前队列里的command。这个方法，解决了并发冲突的问题。但是却带来了另一个问题，就是热点数据的情况。

假如针对某个聚合根的command非常多。那可想而知，某个queue里的command可能都是针对该聚合根的command了。那这样的话这个queue中就基本没机会处理其他聚合根的command，导致其他聚合根的command的处理会被大大延迟。虽然，这种方式遵守了先到先处理的原则，但在这种热点数据的情况下，非热点的数据基本没机会被处理了。设想，现在总共有100个queue可以用来路由，然后现在正好有100个热点修改的聚合根，然后它们的command分别会被路由到对应的queue。这样，我们不难理解，这100个queue中的command就大部分都被这100个聚合根的command占满了。这样导致的结果就是不热点的聚合根的command可能会很晚才会被处理。那怎么样做可以更好呢？

学过akka框架的人应该知道，akka中每个actor都有一个mailbox，mailbox就是一个queue，mailbox中存放所有需要被当前actor处理的消息。如果我们把actor理解为DDD中的聚合根的话，那就是我们可以为每个聚合根设计一个queue，只要是对同一个聚合根的command，都会被先放入queue。然后立即通知任务调度服务，处理当前的聚合根的mailbox。如果当前线程池有可用的线程，那就会立即处理当前的mailbox。如果当前的mailbox正好有command正在被处理中，那本次处理直接结束。然后mailbox中的每个command，在被处理完之后，如果存在下一个command，则会立即请求任务调度服务处理下一个command。

通过这样的设计，我们本质上是为每个聚合根分配一个mailbox，即一个queue；然后，只要是对某个聚合根的command，总是先进入对应的mailbox。然后立即调用任务调度服务去处理当前的mailbox。这个设计，其实也是先到先处理的方式，因为先到的command，会先通知任务调度服务处理。但是不同的是，假如任务调度服务在处理某个mailbox时，如果当前mailbox有command正在被处理，那是会直接结束本次处理的。这样的好处是，任务调度服务可以快速的去处理下一个任务。而原来那种纯粹只设计固定的N个queue的方式，处理command的顺序只能是先进先处理了。

关于mailbox的设计，还有另外一点需要注意的。就是假如mailbox中当前某个command处理完了，然后后面也没有需要处理的command了，我们可以直接将当前mailbox标记为空闲状态吗？不可以。因为我们整个设计是无锁的，也就是说，当我们判断当前mailbox，发现没有后续command需要处理，然后在将mailbox标记为空闲状态前，可能正好又有一个新的command进入了mailbox，且我们可能立即调度了一个任务去处理该mailbox，但是由于当前的mailbox还是在忙的状态，所以就直接结束了。这样导致的后果是，前一个command的处理线程认为当前没有下一个command需要处理；而新进来的command的处理线程认为当前的mailbox还在忙，所以也不处理当前command。所以最后导致这个command一直无法被处理了。直到再下一个command进入mailbox后才正常。虽然这个概率非常低，但在高并发的情况下，完全是很有可能的。

为了解决这个问题，我们可以这样做：在判断是否还有下一个command需要处理后，如果没有，则先将mailbox的状态标记为空闲。然后再判断是否有需要处理的command，如果还是没有，才是真正的没有。如果又有了，则应该立即通知任务调度服务处理当前的mailbox。

下面，我们再回到前面说的N个固定队列的设计思路。其实这个设计还有另一个问题，就是由于每个队列只有一个线程在处理。也就是说，如果我们有100个队列，且队列的消费者是一个个处理command的。那就意味着某一个时刻，最多只能有100个领域事件产生；而对于持久化领域事件，我在之前的文章中有提到，可以通过group commit技术来提高持久化的性能。因为如果每个领域事件的持久化都访问一次DB，那对DB的IO会太频繁。最后会导致我们的架构的瓶颈会被卡在领域事件的持久化上。所以，这种情况下，我们如果用了group commit技术，那一次只能最多持久化100个事件。而且一般是达不到100的，因为当第一批100个事件一次性group commit持久化完成后，这100个线程才能去各自的队列里拿下一个command来处理。假设我们group commit的定时间隔是10ms，那我们是很难在这10ms内又立即产生下一个100个事件的。为什么会这样？估计各位看官是不是看的有点晕了，没关系，其实我也快晕了，呵呵。

因为我们要保证在同一个队列里的command的处理一定是按次序一个个处理的。也就是说，当前一个command没处理完时，是不能处理下一个command的。否则就会出现我文章最开始提到的并发冲突的问题。所以，当当前的command产生事件，并通知事件持久化服务去持久化这个事件时，必须等在那里，直到事件成功持久化为止。

那么有没有办法，既能保证1）对同一个聚合根的修改是串行的（解决并发冲突问题），2）但是聚合根之间是并行的，3）且我们可以完美的配合group commit技术呢？有，就是上面的mailbox的设计。首先，mailbox保证了对同一个聚合根的所有command都是排队的；其次，mailbox能保证它里面的所有的command的处理也是按次序的，也就是前一个command处理完之后才会处理下一个command。第三，对mailbox的处理，完全是交给任务调度服务来完成。也就是mailbox不关系自己被哪个线程处理，它只保证自己内部的command的处理顺序即可。

假如我们的任务调度服务背后是一个最大有200个线程的线程池。那在高并发的情况下，意味着任意时刻，这个线程池中的200个线程都在工作，且每个线程都在处理一个command，且这些线程不需要等待当前自己处理的command产生的事件的持久化。它们只需要负责把事件丢到一个全局唯一的事件队列即可。然后就可以开始处理下一个command handle task了。然后，这个全局唯一的事件队列只有一个消费者线程，它每隔一定时间（比如20ms）持久化一次队列里的事件。一次持久化多少个我们可以自己配置，通过SqlBulkCopy，我们可以达到很好的批量插入事件的效果。注意，这个配置值以及定时持久化的间隔值，不是拍脑袋想出来的，而是需要我们通过实测来得到。

由于，command handle task thread不会等待当前事件的持久化完成，所以它就可以被线程池回收，然后去处理下一个command handle task了。然后，假如某批事件被持久化完成了，则先将这些事件对应的command都标记为完成。比如，调用command的complete方法，complete方法内，可以进一步通知它所属的mailbox去处理下一个可能存在的command。这一批command都通知完成后，就可以立即处理下一批要持久化的事件了。此时，我们知道，在高写入的场景下，一定已经有足够数量的下一批事件了，因为之前的command handle task thread早就已经去处理其他的command了。

所以，通过上面的设计，我们可以在解决command并发冲突的前提下，保证command产生的事件的批量持久化，且能做到每次持久化都有足够多的事件可以被批量持久化。从而整体提高command的处理吞吐量。当然，如果我们的系统写入数据的并发并不高，那是没必要使用group commit技术的。因为group commit毕竟是定时触发的，比如20ms一次。当然，group commit并不是一定要等待20ms才做一次，我们可以在前一次做完后，立即判断接下来是否有事件需要被持久化，如果有，则立即开始下一次批量持久化。

另外一点需要讨论的是，有些command是没有指定聚合根ID的，比如有些新增聚合根的command，可能聚合根ID没有在command上指定，而是在实例化聚合根时才产生。或者，还有一些command不是操作聚合根，而是可能调用外部系统的API。这种command，我们可以直接调用任务调度服务去处理即可。因为这种command不会产生对我们的系统内的聚合根的修改的并发冲突。

说了这么多的思考文字，看起来比较枯燥，下面我们来看看关键部分的代码吧！

public void Process(ProcessingCommand processingCommand)

{

    if (string.IsNullOrEmpty(processingCommand.AggregateRootId))

    {

        _commandScheduler.ScheduleCommand(processingCommand);

    }

    else

    {

        var commandMailbox = _mailboxDict.GetOrAdd(processingCommand.AggregateRootId,

            new CommandMailbox(_commandScheduler, _commandExecutor, _loggerFactory));

        commandMailbox.EnqueueCommand(processingCommand);

        _commandScheduler.ScheduleCommandMailbox(commandMailbox);

    }

}

当一个command被处理时，我们先判断其是否有聚合根ID，如果没有，则直接交给commandScheduler（command处理任务调度服务）进行处理；如果有，则根据聚合根ID找到其mailbox，然后把当前command放入mailbox，然后通知commandScheduler处理该mailbox。接下来我们看看command mailbox的设计：

public class CommandMailbox

{

    private readonly ConcurrentQueue<ProcessingCommand> _commandQueue;

    private readonly ICommandScheduler _commandScheduler;

    private readonly ICommandExecutor _commandExecutor;

    private readonly ILogger _logger;

    private int _isRunning;

    public CommandMailbox(ICommandScheduler commandScheduler, ICommandExecutor commandExecutor, ILoggerFactory loggerFactory)

    {

        _commandQueue = new ConcurrentQueue<ProcessingCommand>();

        _commandScheduler = commandScheduler;

        _commandExecutor = commandExecutor;

        _logger = loggerFactory.Create(GetType().FullName);

    }

    public void EnqueueCommand(ProcessingCommand command)

    {

        command.SetMailbox(this);

        _commandQueue.Enqueue(command);

    }

    public bool MarkAsRunning()

    {

        return Interlocked.CompareExchange(ref _isRunning, , ) == ;

    }

    public void MarkAsNotRunning()

    {

        Interlocked.Exchange(ref _isRunning, );

    }

    public void CompleteCommand(ProcessingCommand processingCommand)

    {

        _logger.DebugFormat("Command execution completed. cmdType:{0}, cmdId:{1}, aggId:{2}",

            processingCommand.Command.GetType().Name,

            processingCommand.Command.Id,

            processingCommand.AggregateRootId);

        MarkAsNotRunning();

        RegisterForExecution();

    }

    public void RegisterForExecution()

    {

        _commandScheduler.ScheduleCommandMailbox(this);

    }

    public void Run()

    {

        ProcessingCommand currentCommand = null;

        try

        {

            if (_commandQueue.TryDequeue(out currentCommand))

            {

                _logger.DebugFormat("Start to execute command. cmdType:{0}, cmdId:{1}, aggId:{2}",

                    currentCommand.Command.GetType().Name,

                    currentCommand.Command.Id,

                    currentCommand.AggregateRootId);

                ExecuteCommand(currentCommand);

            }

        }

        finally

        {

            if (currentCommand == null)

            {

                MarkAsNotRunning();

                if (!_commandQueue.IsEmpty)

                {

                    RegisterForExecution();

                }

            }

        }

    }

    private void ExecuteCommand(ProcessingCommand command)

    {

        try

        {

            _commandExecutor.ExecuteCommand(command);

        }

        catch (Exception ex)

        {

            _logger.Error(string.Format("Unknown exception caught when executing command. commandType:{0}, commandId:{1}",

                command.Command.GetType().Name, command.Command.Id), ex);

        }

    }

}

command mailbox有一个状态标记，表示当前是否正在处理command。这个状态我们会通过原子锁来更改。为了更好的说明问题，我先把commandScheduler的代码也先贴出来：

public class DefaultCommandScheduler : ICommandScheduler

{

    private readonly TaskFactory _taskFactory;

    private readonly ICommandExecutor _commandExecutor;

    public DefaultCommandScheduler(ICommandExecutor commandExecutor)

    {

        var setting = ENodeConfiguration.Instance.Setting;

        _taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(setting.CommandProcessorParallelThreadCount));

        _commandExecutor = commandExecutor;

    }

    public void ScheduleCommand(ProcessingCommand command)

    {

        _taskFactory.StartNew(() => _commandExecutor.ExecuteCommand(command));

    }

    public void ScheduleCommandMailbox(CommandMailbox mailbox)

    {

        _taskFactory.StartNew(() => TryRunMailbox(mailbox));

    }

    private void TryRunMailbox(CommandMailbox mailbox)

    {

        if (mailbox.MarkAsRunning())

        {

            _taskFactory.StartNew(mailbox.Run);

        }

    }

}

从上面的代码可以看到，当command mailbox被commandScheduler处理时，实际上就是创建了一个task，该task是调用TryRunMailbox方法，该方法先尝试将当前的mailbox标记为运行状态，如果当前已经在运行状态，则不做任何处理；如果标记为运行状态成功，则启动一个任务去调用commandMailBox的Run方法。在Run方法内部，mailbox会尝试取出一个需要被处理的command，如果有，就执行该command；如果没有，则先将mailbox当前状态标记为空闲状态，然后再判断一次mailbox中是否有需要处理的command，如果有，则通知commandScheduler处理自己；否则，不做任何处理。

最后，当一个command处理完成后，它会通知自己所属的mailbox。然后mailbox通知commandScheduler处理自己。如下代码所示：

public class ProcessingCommand

{

    private CommandMailbox _mailbox;

    public void SetMailbox(CommandMailbox mailbox)

    {

        _mailbox = mailbox;

    }

    public void Complete(CommandResult commandResult)

    {

        if (_mailbox != null)

        {

            _mailbox.CompleteCommand(this);

        }

    }

}

为了代码的好理解，我去掉了这个类中一些无关的代码。

好了，通过上面的文字介绍和代码，我基本把我想表达的设计写了出来。有点乱，我发现要把自己所想的东西表达清楚，还真不是一件容易的事情。希望通过这篇文章，能让对ENode有兴趣的朋友更好的理解ENode。