分享一个c#写的开源分布式消息队列equeue

分享一个c#写的开源分布式消息队列equeue

前言

equeue消息队列中的专业术语

Topic

Queue

Producer

Consumer

Consumer Group

Broker

集群消费

广播消费

消费进度(offset)

equeue是什么？

关键问题的思考

1.producer,broker,consumer三者之间如何通信

2.消息如何持久化

3.producer发送消息时的消息路由的细节

4.consumer负载均衡如何实现

5.如何实现实时消息推送

6.如何处理消息消费失败的情况

7.如何解决broker的单点问题

前言

本文想介绍一下前段时间在写enode时，顺便实现的一个分布式消息队列equeue。这个消息队列的思想不是我想出来的，而是通过学习阿里的rocketmq后，自己用c#实现了一个轻量级的简单版本。一方面可以通过写这个队列让自己更深入的掌握消息队列的一些常见问题；另一方面也可以用来和enode集成，为enode中的command和domain event的消息传递提供支持。目前在.net平台，比较好用的消息队列，最常见的是微软的MSMQ了吧，还有像rabbitmq也有.net的client端。这些消息队列都很强大和成熟。但当我学习了kafka以及阿里的rocketmq（早期版本叫metaq，自metaq 3.0后改名为rocketmq）后，觉得rocketmq的设计思想深深吸引了我，因为我不仅能理解其思想，还有其完整的源代码可以学习。但是rocketmq是java写的，且目前还没有.net的client端，所以不能直接使用（有兴趣的朋友可以为其写一个.net的client端），所以在学习了rocketmq的设计文档以及大部分代码后，决定自己用c#写一个出来。

项目开源地址：https://github.com/tangxuehua/equeue，项目中包含了队列的全部源代码以及如何使用的示例。也可以在enode项目中看到如何使用。

EQUEUE消息队列中的专业术语

Topic

一个topic就是一个主题。一个系统中，我们可以对消息划分为一些topic，这样我们就能通过topic，将消息发送到不同的queue。

Queue

一个topic下，我们可以设置多个queue，每个queue就是我们平时所说的消息队列；因为queue是完全从属于某个特定的topic的，所以当我们要发送消息时，总是要指定该消息所属的topic是什么。然后equeue就能知道该topic下有几个queue了。但是到底发送到哪个queue呢？比如一个topic下有4个queue，那对于这个topic下的消息，发送时，到底该发送到哪个queue呢？那必定有个消息被路由的过程。目前equeue的做法是在发送一个消息时，需要用户指定这个消息对应的topic以及一个用来路由的一个object类型的参数。equeue会根据topic得到所有的queue，然后根据该object参数通过hash code然后取模queue的个数最后得到要发送的queue的编号，从而知道该发送到哪个queue。这个路由消息的过程是在发送消息的这一方做的，也就是下面要说的producer。之所以不在消息服务器上做是因为这样可以让用户自己决定该如何路由消息，具有更大的灵活性。

Producer

就是消息队列的生产者。我们知道，消息队列的本质就是实现了publish-subscribe的模式，即生产者-消费者模式。生产者生产消息，消费者消费消息。所以这里的Producer就是用来生产和发送消息的。

Consumer

就是消息队列的消费者，一个消息可以有多个消费者。

Consumer Group

消费者分组，这可能对大家来说是一个新概念。之所以要搞出一个消费者分组，是为了实现下面要说的集群消费。一个消费者分组中包含了一些消费者，如果这些消费者是要集群消费，那这些消费者会平均消费该分组中的消息。

Broker

equeue中的broker负责消息的中转，即接收producer发送过来的消息，然后持久化消息到磁盘，然后接收consumer发送过来的拉取消息的请求，然后根据请求拉取相应的消息给consumer。所以，broker可以理解为消息队列服务器，提供消息的接收、存储、拉取服务。可见，broker对于equeue来说是核心，它绝对不能挂，一旦挂了，那producer，consumer就无法实现publish-subscribe了。

集群消费

集群消费是指，一个consumer group下的consumer，平均消费topic下的queue。具体如何平均可以看一下下面的架构图，这里先用文字简单描述一下。假如一个topic下有4个queue，然后当前有一个consumer group，该分组下有4个consumer，那每个consumer就被分配到该topic下的一个queue，这样就达到了平均消费topic下的queue的目的。如果consumer group下只有两个consumer，那每个consumer就消费2个queue。如果有3个consumer，则第一个消费2个queue，后面两个每个消费一个queue，从而达到尽量平均消费。所以，可以看出，我们应该尽量让consumer group下的consumer的数目和topic的queue的数目一致或成倍数关系。这样每个consumer消费的queue的数量总是一样的，这样每个consumer服务器的压力才会差不多。当前前提是这个topic下的每个queue里的消息的数量总是差不多多的。这点我们可以对消息根据某个用户自己定义的key来进行hash路由来保证。

广播消费

广播消费是指一个consumer只要订阅了某个topic的消息，那它就会收到该topic下的所有queue里的消息，而不管这个consumer的group是什么。所以对于广播消费来说，consumer group没什么实际意义。consumer可以在实例化时，我们可以指定是集群消费还是广播消费。

消费进度(offset)

消费进度是指，当一个consumer group里的consumer在消费某个queue里的消息时，equeue是通过记录消费位置(offset)来知道当前消费到哪里了。以便该consumer重启后继续从该位置开始消费。比如一个topic有4个queue，一个consumer group有4个consumer，则每个consumer分配到一个queue，然后每个consumer分别消费自己的queue里的消息。equeue会分别记录每个consumer对其queue的消费进度，从而保证每个consumer重启后知道下次从哪里开始继续消费。实际上，也许下次重启后不是由该consumer消费该queue了，而是由group里的其他consumer消费了，这样也没关系，因为我们已经记录了这个queue的消费位置了。所以可以看出，消费位置和consumer其实无关，消费位置完全是queue的一个属性，用来记录当前被消费到哪里了。另外一点很重要的是，一个topic可以被多个consumer group里的consumer订阅。不同consumer group里的consumer即便是消费同一个topic下的同一个queue，那消费进度也是分开存储的。也就是说，不同的consumer group内的consumer的消费完全隔离，彼此不受影响。还有一点就是，对于集群消费和广播消费，消费进度持久化的地方是不同的，集群消费的消费进度是放在broker，也就是消息队列服务器上的，而广播消费的消费进度是存储在consumer本地磁盘上的。之所以这样设计是因为，对于集群消费，由于一个queue的消费者可能会更换，因为consumer group下的consumer数量可能会增加或减少，然后就会重新计算每个consumer该消费的queue是哪些，这个能理解的把？所以，当出现一个queue的consumer变动的时候，新的consumer如何知道该从哪里开始消费这个queue呢？如果这个queue的消费进度是存储在前一个consumer服务器上的，那就很难拿到这个消费进度了，因为有可能那个服务器已经挂了，或者下架了，都有可能。而因为broker对于所有的consumer总是在服务的，所以，在集群消费的情况下，被订阅的topic的queue的消费位置是存储在broker上的，存储的时候按照不同的consumer group做隔离，以确保不同的consumer group下的consumer的消费进度互补影响。然后，对于广播消费，由于不会出现一个queue的consumer会变动的情况，所以我们没必要让broker来保存消费位置，所以是保存在consumer自己的服务器上。

EQUEUE是什么？

View Code
上面的代码不多分析了，就是先根据consumer group和topic获取所有的consumer，然后对consumer做排序处理。之所以要做排序处理是为了确保负载均衡时对已有的分配情况尽量不发生改变。接下来就是从本地获取topic下的所有queue，同样根据queueId做一下排序。然后就是调用上面的分配算法计算出当前consumer应该分配到哪些queue。最后调用UpdatePullRequestDict方法，用来对新增或删除的queue做处理。对于新增的queue，要创建一个独立的worker线程，开始从broker拉取消息；对于删除的queue，要停止其对应的work，停止拉取消息。

通过上面的介绍和分析，我们大家知道了equeue是如何实现消费者的负载均衡的。我们可以看出，因为每个topic下的queue的更新是异步的定时的，且负载均衡本身也是定时的，且broker上维护的consumer的信息也不是事实的，因为每个consumer发送心跳到broker不是实时发送的，而是比如每隔5s发送一次。所有这些因为都是异步的设计，所以可能会导致在负载均衡的过程中，同一个queue可能会被两个消费者同时消费。这个就是所谓的，我们只能做到一个消息至少被消费一次，但equeue层面做不到一个消息只会被消费一次。实际上像rocketmq这种也是这样的思路，放弃一个消息只会被消费一次的实现（因为代价太大，且过于复杂，实际上对于分布式的环境，不太可能做到一个消息只会被消费一次），而是采用确保一个消息至少会被消费一次（即at least once）.所以使用equeue，应用方要自己做好对每个消息的幂等处理。

5.如何实现实时消息推送

消息的实时推送，一般有两种做法：推模式（push）和拉模式（pull）。push的方式是指broker主动对所有订阅了该topic的消费者推送消息；pull的方式是指消费者主动到broker上拉取消息；对于推模式，最大的好处就是实时，因为一有新的消息，就会立即推送给消费者。但是有一个缺点就是如果消费者来不及消费，它也会给消费者推消息，这样就会导致消费者端的消息会堵塞。而通过拉的方式，有两种实现：1）轮训的方式拉，比如每隔5s轮训一下是否有新消息，这种方式的缺点是消息不实时，但是消费进度完全由消费者自己把控了；2）开长连接的方式来拉，就是不轮训，消费者和broker之间一直保持的连接通道，然后broker一有新消息，就会利用这个通道把消息发送给消费者。

equeue中目前采用的是通过长连接拉取消息的方式。长连接通过socket长连接实现。但是虽然叫长连接，也不是一直不断开，而是也会设计一个超时的限制，比如一个长连接最大不超过15s，超过15s，则broker发送回复给consumer，告诉consumer当前没有新消息；然后consumer接受到这个回复后，就知道要继续发起下一个长连接来拉取。然后假如在这15s中之内，broker上有新消息了，则broker就能立即主动利用这个长连接通知相应的消费者，把消息传给消费者。所以，可以看出，broker上在处理消费者的拉取消息的请求时，如果当前没有新消息，则会hold住这个socket连接，最多hold 15s，超过15s，则发送返回信息，告诉消费者当前无消息，然后消费者再次发送pull message request过来。通过这样的基于长连接的拉取模式，我们可以实现两个好处：1）消息实时推送；2）由消费者控制消息消费进度；

另外，equeue里还实现了消费者自身的自动限流功能。就是假如当前broker上消息很多，即生产者生产消息的速度大于消费者消费消息的速度，那broker上就会有消息被堆积。那此时消费者在拉取消息时，总是会有新消息拉取到，但是消费者又来不及处理这么多消息。所以equeue框架内置了一个限流（流控，流量控制）的设计，就是可以允许用于配制一个消费者端堆积的消息的上限，比如3000，超过这个数目（可配置），则equeue会让消费者以慢一点的频率拉取消息。比如延迟个多少毫秒（延迟时间可配置）再拉取。这样就简单的实现了流控的目的。

6.如何处理消息消费失败的情况

作为一个消息队列，消费者总是可能会在消费消息时抛出异常，在equeue中这种情况就是消息消费失败的情况。通过上面的消费进度的介绍，大家知道了每个queue对某个特定的consumer group，都有一个唯一的消费进度。实际上，消息被拉取到consumer本地后，可能会被以两种方式消费，一种是并行消费，一种是线性消费。

并行消费的意思是，假如当前一次性拉取过来32个消息，那equeue会通过启动task（即开多线程）的方式并行消费每个消息；

线性消费的意思是，消息是在一个独立的单线程中顺序消费，消费顺序和拉取过来的顺序相同。

对于线性消费，假如前一个消息消费的时候失败了，也就是抛异常了，那该怎么办呢？可能想到的办法是重试个3次，但是要是重试后还是失败呢？总不能因为这个消息而导致后面的消息无法把消费吧？呵呵！对于这种情况，先说一下rocketmq里的处理方式吧：它的做法是，当遇到消费失败的情况，没有立马重试，而是直接把这个消息发送到broker上的某个重试队列，发送成功后，就可以往下消费下一个消息了。因为一旦发送到重试队列，那意味着这个消息就最后总是会被消费了，因为该消息不会丢了。但是要是发送到broker的重试队列也不成功呢？这个？！其实这种情况不大应该出现，如果出现，那基本就是broker挂了，呵呵。

rocketmq中，对于这种情况，那会把这个失败的消息放入本地内存队列，慢慢消费它。然后继续往后消费后面的消息。现在你一定很关心queue的offset是如何更新的？这里涉及到一个滑动门的概念。当一批消息从broker拉取到消费者本地后，并不是马上消费的，而是先放入一个本地的SortedDictionary，key就是消息在queue里的位置，value就是消息本身。因为是一个排序的dictionary，所以key最小的消息意味着是最前面的消息，最大的消息就是最后面的消息。然后不管是并行消费还是线性消费，只要某个消息被消费了，那就从这个SortedDictionary里移除掉。每次被移除一个消息时，总是会返回当前这个SortedDictionary里的最小的key，然后我们就能判断这个key是否和上次比是否前移了，如果是，则更新queue的这个最新的offset。因为每次移除一个消息的时候，总是返回当前SortedDictionary里的最小的key，所以，假如当前offset是3，然后offset为4的这个消息一直消费失败，所以不会被移除，但是offset为5,6,7,8的这些消息虽然都消费成功了，但是只要offset为4的这个消息没有被移除，那最小的key就不会往前移动。这个就是所谓的滑动门的概念了。就好比是在铁轨上一辆在跑的动车，offset的往前移动就好比是动车在不断往前移动。因为我们希望offset总是会不断往前移动，所以不希望前面的某个消费失败的消息让这个滑动门停止移动（即我们总是希望这个最小的key能不断变大），所以我们会想方设法让消费失败的消息能不阻碍滑动门的往前移动。所以才把消费失败的消息放入重试队列。

另外一点需要注意一下：并不是每次成功消费完一个消息，就会立马告诉broker更新offset，因为这样那性能肯定很低，broker也会忙死，更好的办法是先只是在本地内存更新queue的offset，然后定时比如5s一次，将最新的offset更新到broker。所以，因为这个异步的存在，同样也会导致某个消息被重复消费的可能性，因为broker上的offset肯定比实际的消费进度要慢，有5s的时间差。所以，再次强调，应用方必须要处理好对消息的幂等处理！比如enode框架中，对每个command消息，框架内部都做了command的幂等处理。所以使用enode框架的应用，自身无需对command做幂等处理方面的考虑。

上面提到了并行消费和线性消费，其实对于offset的更新来说是一样的，因为并行消费无非是多线程同时从SortedDictionary中移除消费成功的消息，而单线程只是单个线程去移除SortedDictionary中的消息。所以我们要通过锁的机制，保证对SortedDictionary的操作是线程安全的。目前用了ReaderWriterLockSlim来实现对方法调用的线层安全。有兴趣的朋友可以去看一下代码。

最后，也是重点，呵呵。equeue目前还没有实现将失败的消息发回到broker的重试队列。这个功能以后会考虑加进去。

7.如何解决broker的单点问题

这个问题比较复杂，目前equeue不支持broker的master-salve或master-master，而是单点的。我觉得一个成熟的消息队列，为了确保在一个broker挂了的时候，要尽量能确保有其他broker可以接替它，这样才能让消息队列服务器的可靠性。但是这个问题实在太复杂。rocketmq目前实现的也只是master-slave的方式。也就是只要主的master挂了，那producer就无法向broker发送消息了，因为slave的broker是只读的，不能直接接受新消息，slave的broker只能允许被consumer拉取消息。

这个问题，要讨论清楚，需要很多分布式方面的知识。由于篇幅的原因，这里就不做讨论了，实际上我自己也搞不清楚到底该如何设计。希望大牛们多多指点，如何实现broker的高可用哈！