架构设计核心特性

NameServer

路由中心/注册中心，管理Broker，类似于kafka中的zookeeper，负责Broker节点注册，Producer和Consumer发送和消费消息会先去NameServer中查找可用的Broker节点；Broker节点在启动的时候就会去查找配置文件中有哪些NameServer，然后在所有的NameServer中注册自己建立长连接，之后每隔30秒发送一个心跳告诉NameServer当前节点还存活可用；而NameServer也会定时每隔10秒查看列表中哪些Broker长时间没有发送心跳，如果120秒没有发送心跳就会将Broker从列表中移除。

?那么RocketMQ为什么不用zookeeper而选择自己造一个NameServer呢？

（1）因为基于RocketMQ的架构设计，他们仅需要一个轻量级的元数据服务器，只需要保证最终一致性，而不需要像zookeeper一样保持强一致。

（2）使用zookeeper担心服务故障，需要考虑高可用问题，需要维护

为什么NameServer之间不需要同步？如何保持一致？

（1）服务注册（Broker新增）：每个Broker启动时都会查看配置文件有多少NameServer，然后向所有的NameServer注册自己保证NameServer彼此一致

（2）服务剔除（Broker关闭或宕机）：Broker关闭会将自己从所有的NameServer中移除；如果Broker宕机，NameServer会10秒监测一次列表中的Broker是否存活，如果Broker挂了，也会将其移除

（3）路由发现（客户端获取最新的Broker list，初始连接、后续连接）：Producer和Consumer会定期每隔30秒刷新一次本地路由列表，通过重试机制可以解决30秒内连接了宕机Broker的问题

?消费模式

集群模式：消费者组里的消费者会负载消费不同的消息
广播模式：消费者组里的消费者会将消息全部消费一遍，所有消费者消费一样的消息

消费模型

RabbitMQ既支持Pull也支持Push，而Kafka只支持Pull；从接口而言RocketMQ也同时支持Pull和Push，但是实际上RocketMQ的Push不是真正的Push，而是通过Pull来实现的，最终启动了一个pull服务

Message Queue

类似于Kafka中的partition分区，创建topic时需要指定写队列数量和读队列数量；写队列数量决定了有几个message queue，用来划分存储位置，如果没有配置，服务端默认写队列数量是8，而生产端默认是4个；读队列数量决定了消费者有几个线程来消费这些消息，用来做负载。

MessageQueue有三个属性：
????????topic：指定当前message queue是属于topic
????????broker名字：指定存在哪个broker上
????????queue编号：一个topic下有多个queue，所以需要根据编号来寻找

工作原理

生产者

生产者消息发送规则

通过MessageQueueSelector实现类来定义发送规则，有以下三个实现类：

SelectMessageQueueByHash（默认）：发送消息时，API底层默认使用此规则，根据队列id取绝对值和队列数取余数，然后id自增达到轮询效果
??

public class SelectMessageQueueByHash implements MessageQueueSelector {
    public SelectMessageQueueByHash() {}
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        int value = arg.hashCode();
        if (value < 0) {
            value = Math.abs(value);
        }

        value %= mqs.size();
        return (MessageQueue)mqs.get(value);
    }
}

SelectMessageQueueByRandom：随机选择一个队列

public class SelectMessageQueueByRandom implements MessageQueueSelector {
    private Random random = new Random(System.currentTimeMillis());
    public SelectMessageQueueByRandom() {}

    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        int value = this.random.nextInt(mqs.size());
        return (MessageQueue)mqs.get(value);
    }
}

SelectMessageQueueByMachineRoom：返回空，没有实现

也可以自定义发送规则（实现MessageQueueSelector接口）：

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Integer id = (Integer) arg;
        int index = id % mqs.size();
        return mqs.get(index);
    }
}, i);