最近正在研究rocketmq，简单记录下设计的不同

互联网系统中Rpc、服务治理、消息中间件基本都是标配，消息中间件能解耦，削峰，高可用并能间接提供达到最终一致性

消息中间件中，消息消费分为最多一次，至少一次和刚好一次，如果需要实现刚好一次，则系统设计难度增大，系统性能损失增加，权衡利弊，rocket实现的是最少一次，消费端可能会重复接收消息（ACK模式下，ACK消息可能丢失），由消费端幂等消费

为什么不用zk，还是从实际需求出发，Topic路由信息无需在集群之间保持强一致性，最终一致即可，从而减少对zk的依赖和性能的损失

消息存储方面，rocket引入文件组，无限循环使用，commitlog文件每个1G，以第一个偏移值为文件名，为了和consumequeue一致，log中还包含了tag，key等信息便于恢复，顺序写，引入内存映射，相同主题的消息被顺序存储在同一文件中，还提供定时清理等防止过度堆积，利用消费队列文件和索引文件及pagecache等提升读性能，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件, 里面有一部分是存储了tag对应的hashcode，经过对比，符合要求的消息被从commitlog中读取出来，消息在消费前，会对比完整的Message Tag字符串，清除hash冲突造成的误读

消息过滤，基于tag等，在存储设计上基于hash等方式提升过滤效率，可以从Broker或者消费端过滤，broker端过滤可以减少传递到消费端的消息，减少网络损失，消费端过滤可以由消费者任意定义

定时消息，如果要支持任意精度的定时消息消费，必须在消息服务端对消息进行排序，势必带来很大的性能损耗，rocketmq设计不支持任意进度的定时消息，只支持特定延迟级别

客户端支持Push（被推送）、pull（自主控制messagequeue的遍历及消息的读取）两种模式

线程池设计，rocketmq会根据不同的任务类型创建不同的线程池，如果该类型没注册，则由other之类的线程池统一处理

Namesrv之间数据可以不一致，彼此之间互不通信

消息发送端提供容错机制，这个地方之前我就有疑问，为什么在客户端或者消费端获取消息存储meta信息之后，namesrv发现变化后不会通知他们。。。原来是由meta使用端的容错机制来保证高可用，降低namesrv的复杂性

消息的顺序性保证，如果要全局一致，必须单一topic，单一生产者及消费者，清除一切并发，可行性比较低，性能和吞吐量无法接受，结合业务，一般是部分顺序消息，发送端将同一业务ID的消息发送到同一个Message Queue,在消费过程中，不并发处理

CommitLog同步，不是经过netty命令的方式，而是直接TCP连接，效率更高，连接成功后，通过对比master和slave的offset，不断进行同步

从broker获得的消息，因为是提交到线程池里并行执行，很难监控和控制执行状态，RocketMQ定义了一个快照类ProcessQueue来解决

负载均衡或消息分配是在消费者端代码中完成，Consumer从broker处获取全局消息，然后自己做负载均衡，只处理分给自己的部分

跟kafka一样，总的消费者数量不要超过topic的队列数，否则多余的消费者收不到消息

Namesrv本身无状态，其中的Broker，topic等状态信息不会持久存储，都是由各个角色定期上报并存储到内存中

事物消息的实现：发送方向RocketMQ发送“待确认”消息，RocketMQ将收到的“待确认”消息持久化后，向发送方回复消息已经发送成功，发送方开始执行本地事件逻辑，发送方根据本地事件逻辑想RocketMQ发送二次确认，RocketMQ收到commit状态则将第一阶段消息标记为可投递，订阅方将能收到该消息，收到rollback状态则删除第一阶段的消息，如果出现异常，服务器在一段时间后未收到确认消息，则服务器将对“待确认”消息发起回查请求，发送方收到回查请求后通过检查对应消息的本地事件执行结果返回对应的状态，RocketMQ收到后继续处理

服务端接受到新请求后，如果队列没有新消息，并不急于返回，通过一个循环不断查看状态，长轮询的核心是，broker端hold住客户端过来的请求一小段时间，在这个时间内有新消息到达，就利用现有的链接立即返回给消息的consumer，长轮询主动权还是掌握在消费端手中，即使broker消息大量积压，也不会主动推送给消费者

在同步刷盘过程种，有一个设计，避免了任务提交与任务执行的锁冲突，由于避免同步刷盘消费任务与其他消费生产者提交任务直接的锁竞争，GroupCommitService提供读容器与写容器，这两个容器每执行完一次任务后，交互，继续消费任务。

        private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();

        private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();

        public synchronized void putRequest(final GroupCommitRequest request) {

            synchronized (this.requestsWrite) {

                this.requestsWrite.add(request);

            }

            if (hasNotified.compareAndSet(false, true)) {

                waitPoint.countDown(); // notify

            }

        }

        private void swapRequests() {

            List<GroupCommitRequest> tmp = this.requestsWrite;

            this.requestsWrite = this.requestsRead;

            this.requestsRead = tmp;

        }