分布式开放消息系统(RocketMQ)的原理与实践

分布式消息系统做为实现分布式系统可扩展、可伸缩性的关键组件，须要具备高吞吐量、高可用等特色。而谈到消息系统的设计，就回避不了两个问题：java

消息的顺序问题

消息的重复问题

RocketMQ做为阿里开源的一款高性能、高吞吐量的消息中间件，它是怎样来解决这两个问题的？RocketMQ 有哪些关键特性？其实现原理是怎样的？web

关键特性以及其实现原理

1、顺序消息

消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了 3 条消息，分别是订单建立、订单付款、订单完成。消费时，要按照这个顺序消费才有意义。但同时订单之间又是能够并行消费的。算法

假如生产者产生了2条消息：M一、M2，要保证这两条消息的顺序，应该怎样作？你脑中想到的多是这样：服务器

你可能会采用这种方式保证消息顺序网络

M1发送到S1后，M2发送到S2，若是要保证M1先于M2被消费，那么须要M1到达消费端后，通知S2，而后S2再将M2发送到消费端。负载均衡

这个模型存在的问题是，若是M1和M2分别发送到两台Server上，就不能保证M1先达到，也就不能保证M1被先消费，那么就须要在MQ Server集群维护消息的顺序。那么如何解决？一种简单的方式就是将M一、M2发送到同一个Server上：分布式

保证消息顺序，你改进后的方法ide

这样能够保证M1先于M2到达MQServer(客户端等待M1成功后再发送M2)，根据先达到先被消费的原则，M1会先于M2被消费，这样就保证了消息的顺序。svg

这个模型，理论上能够保证消息的顺序，但在实际运用中你应该会遇到下面的问题：性能

网络延迟问题

只要将消息从一台服务器发往另外一台服务器，就会存在网络延迟问题。如上图所示，若是发送M1耗时大于发送M2的耗时，那么M2就先被消费，仍然不能保证消息的顺序。即便M1和M2同时到达消费端，因为不清楚消费端1和消费端2的负载状况，仍然有可能出现M2先于M1被消费。如何解决这个问题？将M1和M2发往同一个消费者便可，且发送M1后，须要消费端响应成功后才能发送M2。

但又会引入另一个问题，若是发送M1后，消费端1没有响应，那是继续发送M2呢，仍是从新发送M1？通常为了保证消息必定被消费，确定会选择重发M1到另一个消费端2，就以下图所示。

保证消息顺序的正确姿式

这样的模型就严格保证消息的顺序，细心的你仍然会发现问题，消费端1没有响应Server时有两种状况，一种是M1确实没有到达，另一种状况是消费端1已经响应，可是Server端没有收到。若是是第二种状况，重发M1，就会形成M1被重复消费。也就是咱们后面要说的第二个问题，消息重复问题。

回过头来看消息顺序问题，严格的顺序消息很是容易理解，并且处理问题也比较容易，要实现严格的顺序消息，简单且可行的办法就是：

保证生产者 - MQServer - 消费者是一对一对一的关系

可是这样设计，并行度就成为了消息系统的瓶颈(吞吐量不够)，也会致使更多的异常处理，好比：只要消费端出现问题，就会致使整个处理流程阻塞，咱们不得不花费更多的精力来解决阻塞的问题。

但咱们的最终目标是要集群的高容错性和高吞吐量。这彷佛是一对不可调和的矛盾，那么阿里是如何解决的？

世界上解决一个计算机问题最简单的方法：“刚好”不须要解决它！

有些问题，看起来很重要，但实际上咱们能够经过合理的设计或者将问题分解来规避。若是硬要把时间花在解决它们身上，其实是浪费的，效率低下的。从这个角度来看消息的顺序问题，咱们能够得出两个结论：

一、不关注乱序的应用实际大量存在

二、队列无序并不意味着消息无序

最后咱们从源码角度分析RocketMQ怎么实现发送顺序消息。

通常消息是经过轮询全部队列来发送的(负载均衡策略)，顺序消息能够根据业务，好比说订单号相同的消息发送到同一个队列。下面的示例中，OrderId相同的消息，会发送到同一个队列：

// RocketMQ默认提供了两种MessageQueueSelector实现：随机/Hash

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {

@Override

public MessageQueue select(List mqs, Message msg, Object arg) {

Integer id = (Integer) arg;

int index = id % mqs.size();

return mqs.get(index);

}

}, orderId);

在获取到路由信息之后，会根据MessageQueueSelector实现的算法来选择一个队列，同一个OrderId获取到的队列是同一个队列。

private SendResult send() {

// 获取topic路由信息

TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());

if (topicPublishInfo != null && topicPublishInfo.ok()) {

MessageQueue mq = null;

// 根据咱们的算法，选择一个发送队列

// 这里的arg = orderId

mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);

if (mq != null) {

return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);

}

}

}

2、消息重复

上面在解决消息顺序问题时，引入了一个新的问题，就是消息重复。那么RocketMQ是怎样解决消息重复的问题呢？仍是“刚好”不解决。

形成消息的重复的根本缘由是：网络不可达。只要经过网络交换数据，就没法避免这个问题。因此解决这个问题的办法就是不解决，转而绕过这个问题。那么问题就变成了：若是消费端收到两条同样的消息，应该怎样处理？

一、消费端处理消息的业务逻辑保持幂等性

二、保证每条消息都有惟一编号且保证消息处理成功与去重表的日志同时出现

第1条很好理解，只要保持幂等性，无论来多少条重复消息，最后处理的结果都同样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID，若是新到的消息ID已经在日志表中，那么就再也不处理这条消息。

咱们能够看到第1条的解决方式，很明显应该在消费端实现，不属于消息系统要实现的功能。第2条能够消息系统实现，也能够业务端实现。正常状况下出现重复消息的几率不必定大，且由消息系统实现的话，确定会对消息系统的吞吐量和高可用有影响，因此最好仍是由业务端本身处理消息重复的问题，这也是RocketMQ不解决消息重复的问题的缘由。

RocketMQ不保证消息不重复，若是你的业务须要保证严格的不重复消息，须要你本身在业务端去重。