消息中间件和RPC(HSF)异同

RPC适用场景

• 调用时双向的
• 调用方需要依赖多个服务提供结果

消息队列适用场景

• 消息发送不受限于消息消费方处理速度
• 发送方通过queue进行消息堆积，防止击穿下游服务
• 多个下游服务可以订阅同一个topic进行处理
• 处理耗时操作，比如文件复制，网络下载等等。

概念和术语

消息生产者

简称producer，负责消息产生并发送到meta服务器

消息消费者

简称consumer，负责消息的消费，meta采用pull模型，由消费者主动从meta服务器拉取数据并解析成消息消费。

Topic

消息的主题，由用户定义并在服务端配置。producer发送消息到某个topic下，consumer从某个topic下消费消息

分区（partition）

为了集群和负载均衡，同一个topic下面还分为多个分区，如meta-test这个topic我们可以分为10个分区，分别有两台服务器提供，那么可能每台服务器提供5个分区，假设服务器id分别为0和1，则所有分区为0-0、0-1、0-2、0-3、0-4、1-0、1-1、1-2、1-3、1-4。

Message

消息，负载用户数据，并在生产者、服务端和消费者之间传输

Broker

meta的服务端或者说是服务器，在消息中间件中通常称为broker

消费者分组(Group)

消费者可以是多个消费者共同消费一个topic下的消息，每个消费者消费部分消息。这些消费者就组成一个分组，拥有同一个分组名称,通常也称为消费者集群

Offset

消息在broker上的每个分区都是组织成一个文件列表，消费者拉取数据需要知道数据在文件中的偏移量，这个偏移量就是所谓offset。Offset是绝对偏移量，服务器会将offset转化为具体文件的相对偏移量。详细内容参见#消息的存储结构

MetaQ架构

NameServer

服务的发现，维护了Broker的地址列表和Topic及Topic对应队列的地址列表，与每一个broker保持心跳连接，检车broker是否存活，在producer和consumer需要发布或者消费消息的时候，想nameserver发出请求获取连接。

Broker

broker节点是MetaQ消息存储的位置，所有的消息都由节点broker负责存储。broker会与nameserver建立长连接，将自身的相关信息（比如ip、topic、分区数等）发送到nameserver中，由nameserver集群对节点集群的相应信息进行维护。

Producer

生产者发消息前需要与nameserver建立长连接，生产者将从nameserver上获取到的相关broker节点的信息（比如broker上的topic类型，分区大小等）保存在本地。生产者根据自身需要发送的消息topic类型，从本地broker缓存文件中获取到节点列表，并从中选择某个broker节点，得到节点的相应address信息，然后生产者与这个节点建立连接。如果需要发送的topic在本地缓存中找不到对应的broker，生产者则会根据自身启动时初始化的remotingClient变量（记录了服务器的相关ip和端口），对nameserver进行请求。将自身所提供的topic信息注册到对应的broker节点上。如果注册成功，则最终会返回一个topicPublishInfo，其中记录了存有topic的对应broker列表信息。

生产者产生的消息体的主要由四部分构成：topic、tag、key、msgBody。

• topic：消息的主题。发布和订阅消息是，都以topic为区分标准

• tag：可看做topic的二级分类。一般可以用来对消息进行过滤处理，以获取到更精确的消息。

• key：消息的唯一标识符。用于标识一个消息。当消息的传输和存储过程中出现故障问题时，可通过key来对失败的消息进行快速定位与查找，对于后期维护以及排查错误是一个关键的变量。

• msgBody：消息体。消息的主体部分一般是byte类型数组，在java开发过程中，往往需要通过序列化和压缩等方式对其进行组装。

Consumer

同生产者一样，消费者需要与nameserver建立长连接，消费者将从nameserver上获取到的相关broker节点的信息（比如broker上的topic类型、节点数量、分区大小等）保存在本地。

消费者在进行消息获取时有两种方式：第一种是pull模型，第二种是push模型。

• **pull模型：**即消费者主动与broker节点进行连接通信，然后根据自己所需要的topic消息类型，从broker上拉取下来指定数量的消息。主动权在消费者手中，pull方式的循环间隔不好设定，间隔设定太短，处在“忙等”状态，浪费资源，间隔太长，消息不能及时处理

• **push模型：**broker节点在收到生产者消息后，主动将消息推送到consumer上。Push方式的实时性比较高，但是会加大server端的工作压力，而且由于client的处理能力不能，client不能受server控制。

metaQ采用长轮询pull方式，既解决了pull实时性不够的问题，又不至于大量浪费资源。

消息存储

MetaQ的存储方式采用物理队列+逻辑队列的形式。

物理队列：

一台机器只有一个，也就是本地的文件系统（图中的commit log），存储着实际的数据文件。MetaQ将消息存储在本地文件中，每个文件最大大小为1G，如果写入新的消息时，超过当前文件大小，则会自动新建一个文件。文件名称为起始字节大小。以起始字节大小命名并排序这些文件是有诸多好处的，当消费者要抓取某个起始偏移量开始位置的数据，会变的很简单，只要根据传上来的offset二分查找文件列表，定位到具体文件，然后将绝对offset减去文件的起始节点转化为相对offset，即可找到对应的数据。

刷盘

commit log是以append的方式编写的，保证了顺序写磁盘，顺序写磁盘效率比随机写内存还高，高吞吐量的保证

• 同步刷盘：节点收到消息之后，会立刻把消息写入到磁盘中
• 异步刷盘：节点收到消息之后，并不会第一时间把消息写到磁盘中，而是先写到内存中。当收到N条消息或者经过一段时间后会在统一把消息写入到磁盘中。

读盘

metaQ的所有消息都是持久化的，先写入系统的PageCache(页高速缓存)，然后刷盘，可以保证内存和磁盘都有一份数据，访问时，直接从内存中获取。

[外链图片转存失败(img-fuGJtODu-1562040940329)(C:\Users\xiaozhan.fc\AppData\Roaming\Typora\typora-user-images\1561026231679.png)]

逻辑队列

一台机器可以有多个(topicA_3 topicA-4 等等)，逻辑队列中的存储的是索引文件。服务器将消息存储到文件后，会将该消息在文件的物理位置，消息大小，消息类型封装成一个固定大小的数据结构，暂且称这个数据结构为索引单元吧，大小固定为 16 byte，消息在物理文件的位置称为offset，8个字节，消息size占4个字节，MessageType占4个字节。多个索引单元组成了一个索引文件，索引文件默认固定大小为 20M，和消息文件一样，文件名是起始字节位置，写满后，产生一个新的文件。metaq对于客户端展现的是逻辑队列就是消费队列，consumer从消费队列里顺序取消息进行消费。

这种设计是把物理和逻辑分离，消费队列更加轻量化。所以metaq可以支撑更多的消费队列数，提升消息的吞吐量，并且有一定的消息堆积能力。但是也有缺点：虽然是顺序写入，但是读却是随机读的。

可靠性 顺序 重复

可靠性

生产者可靠性

消息生产者发送消息后返回SendResult，如果isSuccess返回true，则表示消息已经确认发送到服务器，并且被服务器接受存储。整个发送过程是一个同步的过程，保证消息送到服务器并且返回结果。

服务器可靠性

1. 收到消息之后，写入磁盘，写入成功之后，返回应答给生产者

2. os对系统有缓冲：

   • 每1000条（可配置），即强制调用一次force来写入磁盘设备。

   • 每隔10秒（可配置），强制调用一次force来写入磁盘设备。

因此,Meta通过配置可保证在异常情况下（如磁盘掉电）10秒内最多丢失1000条消息。当然通过参数调整你甚至可以在掉电情况下不丢失任何消息。

消费者可靠性保证

消费者是一条一条消费信息。

如果消费某条信息失败（如异常），则会尝试重试消费这条消息（默认最大次数5次）

超过最大次数无法消费，则将消息存储在本地磁盘，由后台线程继续重试。主线程往后走，继续消费。只有在MessageListener确认成功消费一条消息后，meta的消费者才会继续消费另一条消息。由此来保证消息的可靠消费。

存储方案，offset存储

• zookeeper，默认存储在zoopkeeper上，zookeeper通过集群来保证数据安全性
• mysql，可以连接到使用的mysql数据库，只要建立一张特定的表来存储。完全由数据库来保证数据可靠性
• file，文件存储，将offset信息存储在消费者的本地文件中。

顺序

默认处理原则：谁先到达服务器并写入磁盘，则先处理谁。因为消费者针对每个分区都是按照从前到后递增offset的顺序拉取消息。

Meta可以保证，在单线程内使用该producer发送的消息按照发送的顺序达到服务器并存储，并按照相同顺序被消费者消费，前提是这些消息发往同一台服务器的同一个分区。

public interface PartitionSelector {
   
    
    public Partition getPartition(String topic, List<Partition> partitions, Message message) throws MetaClientException;
}

消息重复

消息重复发生的例子：

生产者：生产者发送消息，等待服务器应答，这个时候发生网络故障，服务器实际上已经将消息写入成功，但是由于网络故障没有返回应答，那么生产者会重发，这个时候服务器收到了两条相同的信息。

这种由故障引起的重复，meta是无法避免的，不判断消息的data是否一致，meta仅仅作为载荷来传输

针对消费者来说也会有这个问题，消费的时候，机器突然断电，没有及时将前进后的offset存储起来，则下次启动的时候或者其他同个分组的消费者owner到这个分区的时候，会重复消费这条消息。

消息生产者(Producer)

首先会用到消息回话工厂类——MessageSessionFactory，这个工厂作用是创建生产者和消费者。创建生产者代码：

final MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(initMetaConfig());
        // create producer,强烈建议使用单例
final MessageProducer producer = sessionFactory.createProducer();

点进去 MetaMessageSessionFactory之后，看看这个工厂帮助我们做了哪些事情。

public MetaMessageSessionFactory(final MetaClientConfig metaClientConfig) throws MetaClientException {
   
    
        super();
        try {
   
    
            this.checkConfig(metaClientConfig);
            this.metaClientConfig = metaClientConfig;
            final ClientConfig clientConfig = new ClientConfig();
            clientConfig.setTcpNoDelay(TCP_NO_DELAY);
            clientConfig.setMaxReconnectTimes(MAX_RECONNECT_TIMES);
            clientConfig.setWireFormatType(new MetamorphosisWireFormatType());
            clientConfig.setMaxScheduleWrittenBytes(MAX_SCHEDULE_WRITTEN_BYTES);
            try {
   
    
                this.remotingClient = new RemotingClientWrapper(RemotingFactory.connect(clientConfig));
            }
            catch (final NotifyRemotingException e) {
   
    
                throw new NetworkException("Create remoting client failed", e);
            }
            // 如果有设置，则使用设置的url并连接，否则使用zk发现服务器
            if (this.metaClientConfig.getServerUrl() != null) {
   
    
                this.connectServer(this.metaClientConfig);
            }
            else {
   
    
                this.initZooKeeper();
            }

            this.producerZooKeeper =
                    new ProducerZooKeeper(this.metaZookeeper, this.remotingClient, this.zkClient, metaClientConfig);
            this.sessionIdGenerator = new IdGenerator();
            // modify by wuhua
            this.consumerZooKeeper = this.initConsumerZooKeeper(this.remotingClient, this.zkClient, this.zkConfig);
            this.zkClientChangedListeners.add(this.producerZooKeeper);
            this.zkClientChangedListeners.add(this.consumerZooKeeper);
            this.subscribeInfoManager = new SubscribeInfoManager();
            this.recoverManager = new RecoverStorageManager(this.metaClientConfig, this.subscribeInfoManager);
            this.shutdownHook = new Thread() {
   
    

                @Override
                public void run() {
   
    
                    try {
   
    
                        MetaMessageSessionFactory.this.isHutdownHookCalled = true;
                        MetaMessageSessionFactory.this.shutdown();
                    }
                    catch (final MetaClientException e) {
   
    
                        log.error("关闭session factory失败", e);
                    }
                }

            };
            Runtime.getRuntime().addShutdownHook(this.shutdownHook);
        }
        catch (MetaClientException e) {
   
    
            this.shutdown();
            throw e;
        }
        catch (Exception e) {
   
    
            this.shutdown();
            throw new MetaClientException("Construct message session factory failed.", e);
        }
    }

可以看到，这个工厂大致帮我们做了下面几个事情：

1. 服务的查找和发现，通过diamond和zookeeper帮你查找日常的meta服务器地址列表

2. 连接的创建和销毁，自动创建和销毁到meta服务器的连接，并做连接复用，也就是到同一台meta的服务器在一个工厂内只维持一个连接。

3. 消息消费者的消息存储和恢复。

4. 协调和管理各种资源，包括创建的生产者和消费者的。

消息

消息属性

消息是邮件，MetaQ是邮局，消息填好目的地之后，放在邮局，邮局就可以帮助你送到目的地点
MetaQ的消息在Java客户端里是com.taobao.metamorphosis.Message类，它主要包括这么几个属性：

public class Message implements Serializable {