另外，上面我说过EQueue中目前有两种Chunk文件，一种是存储消息本身的，我叫做Message Chunk；一种是存储队列信息的，我叫做Queue Chunk；而Queue Chunk的数据是依赖于Message Chunk的。上面我说的两种删除策略是针对Message Chunk而言的。而Queue Chunk，由于这个依赖性，我觉得比较合理的方式是，只需要判断当前Queue Chunk中的所有的消息对应的Message Chunk是否已经都删除了，如果是，难说明这个Queue Chunk也已经没意义了，就可以删除了。但只要这个Queue Chunk中至少还有一个消息的Chunk文件没删除，那这个Queue Chunk就不会删除。
上面这个只是思路哦，真实的代码肯定比这个复杂，呵呵。有兴趣的朋友还是要看代码的。
如何查消息？之前用SQL Server的方式，由于DB很容易查消息，所以查询消息不是大问题。但是现在我们的消息是放在文件里的，那要怎么查询呢？目前支持根据消息ID来查询。当Producer发送一个消息到Broker，Broker返回结果里会包含消息的ID。Producer的正确做法应该是要用日志或其他方式记录这个ID，并最好和自己的当前业务消息的某个业务ID一起记录，比如CommandId或者EventId，这样我们就能根据我们的业务ID找到消息ID，然后根据消息ID找到消息内容了。那消息ID现在是怎么设计的呢？也是受到RocketMQ的启发，消息ID由两部分组成：1）Broker的IP；2）消息在Broker的文件中的全局位置；这样，当我们要根据某个消息ID查询时，就可以先定位到这个消息在哪个Broker上，也同时知道了消息在文件的哪个位置了，这样就能最终读取这个消息的内容了。
为什么要这样设计呢？如果我们的Broker没有集群，那其实不需要包含Broker的IP；这个设计是为了未来EQueue Broker会支持集群的，那个时候，我们就必须要知道某个消息ID对应的Broker是哪个了。
如何保存队列消费进度？EQueue中，每个Queue，都会有一个对应的Consumer。消费进度就是这个Queue当前被消费到哪里了，一个Offset值。比如Offset为100，就表示当前这个Queue已经消费到第99（因为是从0开始的）个位置的消息了。因为一个Broker上有很多的Queue，比如有100个。而我们现在是使用文件的方式来存储信息了。所以自然消费进度也是用文件了。但由于消费进度的信息很少，也不是递增的形式。所以我们可以简单设计，目前EQueue采用一个文件的方式来保存所有Queue的消费进度，文件内容为JSON，JSON里记录了每个Queue的消费进度。文件内容看起来像下面这样：
{"SampleGroup":{"topic1-3":89055,"topic1-2":89599,"topic1-1":89471,"topic1-0":89695}}
上面的JSON标识一个名为SampleGroup的ConsumerGroup，他消费了一个名为topic1的topic，然后这个topic下的每个Queue的消费进度记录了下来。如果有另一个ConsumerGroup，也消费了这个topic，那消费进度是隔离的。如果还不清楚ConsumerGroup的同学，要去看一下我之前写的EQueue的文章了。
还有没有可以优化的地方？到目前为止，还有没有其他可优化的大的地方呢？有。之前我做EQueue时，总是把消息从数据库读取出来，然后构造出消息对象，再把消息对象序列化为二进制，再返回给Consumer。这里涉及到从DB拿出来，再序列化为二进制。学习了RocketMQ的代码后，我们可以做的更聪明一点。因为其实基于文件存储时，我们从文件里拿出来的已经是二进制了。所以可以直接把二进制返回给消费者即可。不需要先转换为对象再做序列化了。通过这个设计的改进，我们现在的消费者消费消息，可以说无任何瓶颈了，非常快。
如何统计消息读写情况？在测试写文件的这个版本时，我们很希望知道每个Chunk的读写情况的统计，从而确定设计是正确的。所以，我给EQueue的Chunk增加了实时统计Chunk读写的统计服务。目前我们在运行EQueue自带的例子时，Broker会每个一秒打印出所有Chunk的读写情况，这个特性极大的方便我们判断消息的发送和消费是否正常，消费是否有延迟等。
其他新增功能 更完善和安全的队列扩容和缩容设计这次我给EQueue的Web后台管理控制台也完善了一下队列的增加和减少的设计。增加队列（即队列的扩容）比较简单，直接新增即可。但是当我们要删除一个队列时，怎样安全的删除呢？主要是要确保删除这个队列时，已经没有Producer或Consumer在使用这个队列了。要怎么做到呢？我的思路是，为每个Queue对象设计两个属性，表示对Producer是否可见，对Consumer是否可见。当我们要删除某个Queue时，可以：1）先让其对Producer不可见，这样Producer后续就不会再发送新的消息到这个队列了；然后等待，直到这个队列里的消息都被所有的消费者消费掉了；然后再设置为对Consumer不可见。然后再过几秒，确保每个消费者都不会再向这个队列发出拉取消息的请求了。这样我们就能安全的删除这个队列了。删除队列的逻辑大概如如下：
public void DeleteQueue(string topic, int queueId) { lock (this) { var key = QueueKeyUtil.CreateQueueKey(topic, queueId); Queue queue; if (!_queueDict.TryGetValue(key, out queue)) { return; } //检查队列对Producer或Consumer是否可见，如果可见是不允许删除的 if (queue.Setting.ProducerVisible || queue.Setting.ConsumerVisible) { throw new Exception("Queue is visible to producer or consumer, cannot be delete."); } //检查是否有未消费完的消息 var minConsumedOffset = _consumeOffsetStore.GetMinConsumedOffset(topic, queueId); var queueCurrentOffset = queue.NextOffset - 1; if (minConsumedOffset < queueCurrentOffset) { throw new Exception(string.Format("Queue is not allowed to delete, there are not consumed messages: {0}", queueCurrentOffset - minConsumedOffset)); } //删除队列的消费进度信息 _consumeOffsetStore.DeleteConsumeOffset(queue.Key); //删除队列本身，包括所有的文件 queue.Delete(); //最后将队列从字典中移除 _queueDict.Remove(key); } }
代码应该很简单直接，不多解释了。队列的动态新增和删除，可以方便我们线上应付在线活动时，随时为消费者提供更高的并行消费能力，以及活动结束后去掉多余的队列。是非常实用的功能。
支持Tag功能
相关文章
精彩导读
热门资讯
关注我们
