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。

ENode 2.0 - 介绍一下关于ENode中对Command的调度设计的更多相关文章

  1. ENode 2.0

    ENode 2.0 - 介绍一下关于ENode中对Command的调度设计 摘要: CQRS架构,C端的职责是处理从上层发送过来的command.对于单台机器来说,我们如何尽快的处理command呢? ...

  2. ENode 1.0 - 整体架构介绍

    前言 今天是个开心的日子,又是周末,可以安心轻松的写写文章了.经过了大概3年的DDD理论积累,以及去年年初的第一个版本的event sourcing框架的开发以及项目实践经验,再通过今年上半年利用业余 ...

  3. ENode 2.0 - 整体架构介绍

    前言 今天是个开心的日子,又是周末,可以轻轻松松的写写文章了.去年,我写了ENode 1.0版本,那时我也写了一个分析系列.经过了大半年的时间,我对第一个版本做了很多架构上的改进,最重要的就是让ENo ...

  4. ENode 2.0 - 第一个真实案例剖析-一个简易论坛(Forum)

    前言 经过不断的坚持和努力,ENode 2.0的第一个真实案例终于出来了.这个案例是一个简易的论坛,开发这个论坛的初衷是为了验证用ENode框架来开发一个真实项目的可行性.目前这个论坛在UI上是使用了 ...

  5. ENode 1.0 - 消息的重试机制的设计思路

    项目开源地址:https://github.com/tangxuehua/enode 上一篇文章,简单介绍了enode框架中消息队列的设计思路,本文介绍一下enode框架中关系消息的重试机制的设计思路 ...

  6. ENode 1.0 - Staged Event-Driven Architecture思想的运用

    开源地址:https://github.com/tangxuehua/enode 上一篇文章,简单介绍了enode框架的command service api设计思路.本文介绍一下enode框架对St ...

  7. ENode 1.0 - 框架的总体目标

    开源地址:https://github.com/tangxuehua/enode 本文想介绍一下enode框架要实现的目标以及部分实现分析思路剖析.总体来说enode框架是一个基于cqrs架构和消息驱 ...

  8. 开放平台鉴权以及OAuth2.0介绍

    OAuth 2.0 协议 OAuth是一个开发标准,允许用户授权第三方网站或应用访问他们存储在另外的服务提供者上的信息,而不需要将用户名和密码提供给第三方网站或分享他们数据的内容. OAuth 2.0 ...

  9. Windows PE3.0制作方法(从Win7中提取制作)

    Windows PE3.0制作方法(从Win7中提取制作 在d:新建文件夹winpe,在winpe中新建sources.pe3和new文件夹,把附件中提供的工具imagex连文件夹一起放到winpe目 ...

随机推荐

  1. smarty 时间格式化date_format

    代码如下:$smarty = new Smarty; $smarty->assign('yesterday', strtotime('-1 day')); $smarty->display ...

  2. Leaving Auction

    Leaving Auction 题目链接:http://codeforces.com/contest/749/problem/D 二分 本来以为是哪种神奇的数据结构,没想到sort+lower_bon ...

  3. 在游览器上可以连网,Ionic打包后不能连接网络

    在游览器上可以连网,Ionic打包后不能连接网络.可能是没有安装cordova-plugin-whitelist插件.  解决方案:

  4. Matplotlib初体验

    为一个客户做了关于每个差异otu在时间点上变化的折线图,使用python第一次做批量作图的程序,虽然是很简单的折线图,但是也是第一次使用matplotlib的纪念. ps:在第一个脚本上做了点小的改动 ...

  5. 分类导航菜单的制作(附源码)--HTML

    不多说,直接贴代码哈!有疑问,可追加评论哈! demo.html: <!DOCTYPE html><html> <head> <title>分类导航菜单 ...

  6. Asp .Net Core 读取appsettings.json配置文件

         Asp .Net Core 如何读取appsettings.json配置文件?最近也有学习到如何读取配置文件的,主要是通过 IConfiguration,以及在Program中初始化完成的. ...

  7. 《vue.js快跑》总结:为什么选择VUE

    2019-3-31 为什么选择Vue 有这个一个需求,我们需要根据后端数据接口请求返回的数组在页面中按列表展示? 传统上我们使用jQuery的Ajax发送http请求,获取数据.判断列表数据是否存在, ...

  8. sparkR介绍及安装

    sparkR介绍及安装 SparkR是AMPLab发布的一个R开发包,为Apache Spark提供了轻量的前端.SparkR提供了Spark中弹性分布式数据集(RDD)的API,用户可以在集群上通过 ...

  9. _lottery

    通过积分购买彩票,奖励以积分形式发放 当aaa_chance,max_chance,min_chance均为0时,自动计算系统最小积分开销进行开奖

  10. linux:用户及文件权限管理

    学习内容来自实验楼.莫烦python.CSDN 一.Linux 用户管理 1. 查看用户 who am i 或者who mom likes who -a:打印所有能打印的  who -d :打印死掉的 ...