消息中间件三大特性：异步、削峰、解耦

一、 安装部署

1.安装

官网下载，解压后直接运行（单Master）

#启动namesrv
nohup sh bin/mqnamesrv &
#查看日志
tail -f ~/logs/rocketmqlogs/namesrv.log

#先启动broker
nohup sh bin/mqbroker -n localhost:9876 &
查看日志
tail -f ~/logs/rocketmqlogs/broker.log 

注意Broker的IP设置，手动修改属性 echo "brokerIP1=主机外网IP" > broker.conf
指定配置文件启动：nohup sh bin/mqbroker -n ${namesrvIp}:9876 -c conf/broker.conf
参考：https://github.com/apache/rocketmq#readme

2.部署模型

部署模型图

NameServer

NameServer是一个简单的 Topic 路由注册中心，支持 Topic、Broker 的动态注册与发现， NameServer是无状态化，节点之间无任何信息同步。

主要包括两个功能：

• Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活
• 路由信息管理，每个NameServer将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。

Broker

Broker主要负责消息的存储、投递和查询以及服务高可用保证. Broker采用Master-Slave架构解决高可用问题，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master可以多个，提升消息写入能力

1.每个 Broker 与 NameServer 集群中的所有节点建立长连接，定时注册 Topic 信息到所有 NameServer
2.Producer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取Topic路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态
3.Consumer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的Master、Slave 建立长连接，且定时向 Master、Slave发送心跳。Consumer 既可以从 Master 订阅消息，也可以从Slave订阅消息。

部署方式

• 单Master模式：只有一个节点，一般用于测试环境
• 多Master模式：多个Master，提升负载能力，但是不部署Slave，单台Master故障，该Master上未被消费的消息在机器恢复之前不可消费
• 多Master-Slave异步复制：解决单台Master故障问题，消费者可以从Slave消费，由于是异步，主备有短暂消息延迟（毫秒级）
• 多Master-Slave同步复制: 同上，只是采用主备采用同步机制，性能略低10%左右，发送单个消息RT会略高，但是消息无延迟

一般生产环境采用的是多Master-Slave异步复制模式

二、消息模型

消息模型图

消费者组

为了消息消费能力的水平扩展，引入消费者组。

消费模式

同一个消费者组的消费模式：

• 集群模式（默认）
  同一个 ConsumerGroup 中的 Consumer 实例是负载均衡消费，即任意一条消息只需要被消费组内的任意一个消费者处理即可
• 广播模式
  同一个 ConsumerGroup 中的每个 Consumer 实例都处理全部的队列

1.集群模式下扩缩消费者数量也无法提升或降低消费能力，但当Topic的总队列数小于消费者的数量时，消费者将分配不到队列，即使消费者再多也无法提升消费能力
3.广播模式下扩缩消费者数量也无法提升或降低消费能力

• 支持推Push和拉Pull两种模式

三、基本概念

队列

为了支持高并发和水平扩展，需要对 Topic 进行分区，称为队列。一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上

主题Topic&Tag

Topic 与 Tag 都是业务上用来归类的标识，区别在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。可使用 Tag 可以实现对 Topic 中的消息进行过滤。通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息

Keys

消息的唯一标识码，方便定位消息丢失问题。

消费位点

每个队列都会记录自己的最小位点、最大位点，针对消费者组，会有消费位点，在集群模式下，消费位点是由客户端提给交服务端保存的，在广播模式下，消费位点是由客户端自己保存的

四、代码示例

消息发送分为三种方式：同步、异步、单向传输，前两种是可靠的，无论是否成功都有相应，最后一种只管发送没有返回结果

public class SyncProducer {
  public static void main(String[] args) throws Exception {
    // 初始化一个producer并设置Producer group name
    DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name"); //（1）
    // 设置NameServer地址
    producer.setNamesrvAddr("localhost:9876");  //（2）
    // 启动producer
    producer.start();
    for (int i = 0; i < 100; i++) {
      // 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤
      Message msg = new Message("TopicTest" /* Topic */,
        "TagA" /* Tag */,
        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        );   //（3）
      // 利用producer进行发送，并同步等待发送结果
      SendResult sendResult = producer.send(msg);   //（4）
      System.out.printf("%s%n", sendResult);
    }
    // 一旦producer不再使用，关闭producer
    producer.shutdown();
  }
}

Push消费

public class Consumer {
  public static void main(String[] args) throws InterruptedException, MQClientException {
    // 初始化consumer，并设置consumer group name
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
   
    // 设置NameServer地址 
    consumer.setNamesrvAddr("localhost:9876");
    //设置消费模式，默认是集群模式
    //consumer.setMessageModel(MessageModel.BROADCASTING);
    //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
    consumer.subscribe("TopicTest", "*");
    //注册回调接口来处理从Broker中收到的消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        System.out.println(new String(msgs.get(0).getBody(),StandardCharsets.UTF_8))
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
        // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    // 启动Consumer
    consumer.start();
    System.out.printf("Consumer Started.%n");
  }
}

Pull 消费

public class LitePullConsumerSubscribe {
    public static volatile boolean running = true;
    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer_test");
        litePullConsumer.subscribe("TopicTest", "*");
        litePullConsumer.setPullBatchSize(20);
        litePullConsumer.start();
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                System.out.printf("%s%n", messageExts);
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
}

控制台

rocketmq-dashboard

官网：https://github.com/apache/rocketmq-dashboard

官网：https://rocketmq.apache.org/
Github: https://github.com/apache/rocketmq