RocketMQ_consumer.registermessagelistener

MQ是什么

MQ全称为Message Queue，即消息队列 ，是一种提供消息队列服务的中间件，也称为消息中间件，是一套提供了消息生 产、存储、消费全过程的软件系统，遵循FIFO原则。

MQ的使用场景(重要)

- 限流削峰
  MQ可以将系统的超量请求暂存其中，以便系统后期可以慢慢进行处理，从而避免了请求的丢失或系统 被压垮。
- 异步&解耦
  上游系统对下游系统的调用若为同步调用，则会大大降低系统的吞吐量与并发度，且系统耦合度太高。 而异步调用则会解决这些问题。所以两层之间若要实现由同步到异步的转化，一般性做法就是，在这两层间添加一个MQ层。 即使消费者挂掉也不影响生产者工作，只要把消息放入队列即可，消费者重启后自己消费即可。
- 数据收集
  分布式系统会产生海量级数据流，如：业务日志、监控数据、用户行为等。针对这些数据流进行实时或 批量采集汇总，然后对这些数据流进行大数据分析，这是当前互联网平台的必备技术。通过MQ完成此 类数据收集是最好的选择。
- 大数据处理
  比如我们的平台向“三方平台”获取数据，一次请求了大量数据回来要进行处理，由于数据较多处理不过来，那么就可以放入MQ，再创建一些消费者进行数据处理即可。

【注意】如下情况不太适合MQ

- 小项目，体量不大，并发量低的使用MQ会太过笨重 - 你可以考虑使用Redis做一个消息队列。
- 对数据的一致性有要求(强一致性)的的场景不适合使用MQ，因为MQ是异步的。

使用MQ的好处

- 提高系统响应速度
  任务异步处理。 将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
- 提高系统稳定性
  一是并发被消峰后，系统不容易被高并发打垮，二是系统挂了也没关系,操作内容放到消息队列不丢失，后续重新消费者一样能消费做业务处理。 
- 排序保证 FIFO
  遵循队列先进先出的特点，能够保证消息按照添加的数据被消费。

常见MQ产品(了解)

- ActiveMQ

          ActiveMQ是使用Java语言开发一款MQ产品。早期很多公司与项目中都在使用。但现在的社区活跃度已 经很低。现在的项目中已经很少使用了。 

- RabbitMQ 

  RabbitMQ是使用ErLang语言开发的一款MQ产品。其吞吐量较Kafka与RocketMQ要低，且由于其不是 Java语言开发，所以公司内部对其实现定制化开发难度较大。 

- Kafka

  Kafka是使用Scala/Java语言开发的一款MQ产品。其最大的特点就是高吞吐率，常用于大数据领域的实 时计算、日志采集等场景。其没有遵循任何常见的MQ协议，而是使用自研协议。对于Spring Cloud Netflix，其仅支持RabbitMQ与Kafka。 

- RocketMQ

  RocketMQ是使用Java语言开发的一款MQ产品。经过数年阿里双11的考验，性能与稳定性非常高。其 没有遵循任何常见的MQ协议，而是使用自研协议。对于Spring Cloud Alibaba，其支持RabbitMQ、 Kafka，但提倡使用RocketMQ

RocketMQ

RocketMQ是什么

RocketMQ是一个统一消息引擎、轻量级数据处理平台。

RocketMQ是⼀款阿⾥巴巴开源的消息中间件，双十一承载了万亿级消息的流转，2016年11⽉，阿⾥巴巴向 Apache 软件基⾦会捐赠 RocketMQ，成为 Apache 孵化项⽬，2017 年 9 ⽉ ，Apache 宣布 RocketMQ孵化成为 Apache 顶级项⽬（TLP ）成为国内⾸个互联⽹中间件在 Apache 上的顶级项⽬。

RocketMQ特征(了解)

• 支持集群模型、负载均衡、水平扩展能力

• 亿级别消息堆积能力

• 采用零拷贝的原理，顺序写盘，随机读

• 底层通信框架采用Netty NIO

• NameServer代替Zookeeper，实现服务寻址和服务协调

• 消息失败重试机制、消息可查询

• 强调集群无单点，可扩展，任意一点高可用，水平可扩展

• 经过多次双十一的考验

下载

注意：下载路径不能有中文。C盘路径（包括电脑用户名）和环境变量的CLASSPATH不能有空格

CLASSPATH的空格会有提示，电脑用户名的空格需要改用户名，可以去搜一下

下载地址：

RocketMQ · 官方网站 | RocketMQ (apache.org)

下载后解压

- Bin : 可执行文件目录
- Conif：配置文件目录
- Lib : 依赖库，一堆Jar包

配置

ROCKETMQ_HOME

解压压缩包，配置 ROCKETMQ_HOME

注意不是bin目录，而是跟目录

启动MQ

NameServer

Cmd命令框执行进入至MQ文件夹\bin下，然后执行 start mqnamesrv.cmd，启动NameServer。(注册中心)

启动Boker(MQ)

进入至MQ文件夹\bin下，修改Bean目录下的 runbroker.cmd 中JVM占用内存大小

CMD执行start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true ，启动Broker。

 

RocketMQ插件

为了方便管理，我们需要安装一个可视化插件

下载插件

RocketMQ可视化管理插件下载地址：https://github.com/apache/rocketmq-externals/releases

修改配置

解压后，修改配置：src/main/resource/application.properties ,这里需要指向Name Server 的地址和端口 如下：

 

打包插件

回到安装目录(pom.xml所在目录),执行: mvn clean package -Dmaven.test.skip=true ，然后会在target目录生成打包后的jar文件

 

启动插件

进入 target 目录，执行 java -jar rocketmq-console-ng-1.0.0.jar  ， 访问 http://localhost:8080
 

 可以将三个项目放一起，打包一个启动文件

 一次性启动三个文件

RokcetMQ架构

RocketMQ开发官方文档：

https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

RocketMQ的集群架构如下

 

RocketMQ主要由 Producer、Broker、Consumer、NameServer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器。

为了实现高可用，Broker本身是主备架构，Master负责写请求和读请求，Slave负责数据备份和分担读请求。这样的好处是提高了吞吐量同时防止数据丢失。

Producer（生产者）

消息发布的角色，支持分布式集群方式部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。

RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。

 Producer会使用一定的算法选择把消息发送到哪个master的某个queue中。
 

Consumer（消费者）

消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。Consumer 支持两种消费形式：拉取式消费、推动式消费。（主动，被动），RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息，不同的 Consumer Group可以消费同一个Topic。

Broker

Broker主要负责消息的存储、投递和查询以及服务高可用保证。

NameServer

NameServer是一个Broker与Topic路由的注册中心支持Broker的动态注册与发现主要包括两个功能

• Broker管理

  NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活。

• 路由信息管理

  每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费

Rocket入门
导包

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.8.0</version>
</dependency>

生产者

步骤分析

1. 创建producer组

2. 设置NameServer地址

3. startr生产者

4. 发送消息获取结果

5. 结束producer

public class Producer {
    public static void main(String[] args) throws Exception {
 
        //1.创建生产者组
        DefaultMQProducer producer = new DefaultMQProducer("producer-hello");
        //2.设置NameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //3.启动producer实例
        producer.start();
        //4.创建消息
        Message message = new Message("topic-order", "topic-order-flow", "this is a flow message".getBytes(CharsetUtil.UTF_8));
        //5.发送消息
        producer.send(message);
        //6.关闭producer实例
        System.out.println("消息发送成功");
    }
}

消费者

1. 创建consumer组

2. 设置Name Server地址

3. 设置消费位置，从最开始销毁

4. 设置消息回调处理监听 -> 处理消息

5. Start consumer

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.创建消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-hello");
        //2.设置NameServer地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //3.订阅topic，指定tag标签
        consumer.subscribe("topic-order","topic-order-flow");
        //4.注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                list.forEach(messageExt -> {
                    byte[] body = messageExt.getBody();
                    String msg = new String(body, CharsetUtil.UTF_8);
                    System.out.println("I get it " + msg);
                });
                // 消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者
        consumer.start();
 
    }
}

RabbitMQ工作原理

Topic 消息主题

Topic表示一类消息的集合，每个topic主题包含若干条message消息，每条message消息只能属于一个topic主题，Topic是RocketMQ进行消息订阅的基本单位。

MessageQueue队列

一个Topic中可以包含多个Queue，一 个Topic的Queue也被称为一个Topic中消息的分区（Partition）。 在一个Consumer Group内，一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。这样的分配规则，每个Queue只有一个消费者，可以避免消费过程中的多线程处理和资源锁定，有效提高各Consumer消费的并行度和处理效率。

消息拉取模式

消息的消费分为：拉取式 pull ,和推送是 push 

- Pull:拉取式，需要消费者间隔一定时间就去遍历关联的Queue,实时性差但是便于应用控制消息的拉取
- Push:默认推送式，封装了Queue的遍历，实时性强，但是对系统资源占用比较多。

Queue的分配算法

Queue是如何分配给Consumer的，这对应了四种算法：平均分配策略，环形平均策略，一致性Hash策略，同机房策略。

- 平均分配【默认】：根据 qeueuCount / consumerCount 作为每个消费者平均分配数量，如果多出来的queue就再依次逐个分配给Consumer。
- 环形平均策略：根据消费者的顺序，一个一个的分配Queue即可类似于发扑克牌。
- 一致性Hash策略 ： 该算法将Consumer的Hash值作为节点放到Hash环上，然后将Queue的hash值也放入Hash环上，通过顺时针进行就近分配。
- 同机房策略：该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配。

平均分配性能比较高，一致性Hash性能不高，但是能减少Rebalance，如果Consumer数量变动频繁可以使用一致性Hash。

Offset管理

RockertMQ通过Offset来维护Consumer的消费进度，比如：消费者从哪个位置开始持续消费消息的？这里有三个枚举来指定从什么位置消费

- CONSUME_FROM_LAST_OFFSET：从queue的最后一条消息开始消费 
- CONSUME_FROM_FIRST_OFFSET：从queue的第一条消息开始消费 
- CONSUME_FROM_TIMESTAMP：从某个时间戳位置的消息开始消费。

消费者消费结束之后，会向Consumer会提交其消费进度offset给Broker。Offset信息的存储分为本地 Offset管理 和远程Offset管理

- 远程Offset管理:Brocker通过 store/config/consumerOffset.json 文件以JSON方式来存储offset相关数据以json的形式：适用于集群模式
- 本地Offset管理：offset相关数据以json的形式持久化到Consumer本地磁盘文件中，路径为当前用户主目录下的.rocketmq_offsets/{clientId}/{group}/Offsets.json ：适用于广播模式

Offset的同步提交与异步提交: 集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种： 

- 同步提交：消费者在消费完一批消息后会向broker提交这些消息的offset，等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取 nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费 者是阻塞的。其严重影响了消费者的吞吐量。 
- 异步提交：消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset 后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取 nextBeginOffset。

Rocket使用

导包

<parent>
    <groupId> org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.5.RELEASE</version>
</parent>
 
<dependencies>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
       <!-- <version>2.0.4</version> -->
         <version>2.2.1</version>
    </dependency>
 
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

启动类

@SpringBootApplication
public class ApplicationStart {
    public static void main(String[] args) {
        SpringApplication.run(ApplicationStart.class, args);
    }
}

yaml文件

rocketmq:
  name-server: 127.0.0.1:9876
  #生产者配置
  producer:
    #生产者组名字
    group: "service-producer"
    # 消息最大长度 默认 1024 * 1024 * 4 (4M)
    max-message-size: 4194304
    # 发送消息超时时间，默认 3000
    send-message-timeout: 3000
    # 发送消息失败重试次数，默认2
    retry-times-when-send-failed: 2
    # 异步消息发送失败重试次数
    retry-times-when-send-async-failed: 2
    #达到 4096 ，进行消息压缩
    compress-message-body-threshold: 4096
  consumer:
    #消费者名字
    group: "service-consumer"
    #批量拉取消息数量
    pull-batch-size: 10
    message-model: CLUSTERING
    selector-expression: "*"    
server:
  port: 8888 # 自己随便配的

生产者

@RestController
public class ProducerController {
 
    @Autowired
    private RocketMQTemplate template;
 
    // 同步发送
    @RequestMapping("/send/{message}")
    public String sendMessage(@PathVariable("message")String data){
        Message<String> msg = MessageBuilder.withPayload(data).build();
        template.send("topic-order:topic-order-flow", msg);
        return "send successfully";
    }
 
    // 同步发送 synSend 可以设置超时时间
    @RequestMapping("/sendAgain/{message}")
    public String sendAgain(@PathVariable("message")String data){
        Message<String> msg = MessageBuilder.withPayload(data).build();
        SendResult result = template.syncSend("topic-order:topic-order-flow", msg, 2000);
        System.out.println(result);
        if(result.getSendStatus() == SendStatus.SEND_OK){
            return "synsend successfully";
        } else {
            return "fail to synsend";
        }
    }
 
    // 异步发送
    @RequestMapping("/asynSend/{message}")
    public String asynSend(@PathVariable("message")String data){
        Message<String> msg = MessageBuilder.withPayload(data).build();
        template.asyncSend("topic-order:topic-order-flow",msg,new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
                System.out.println("send successfully");
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("fail to send");
            }
        });
        return "asynsend successfully";
    }
 
    // 单向发送
    @RequestMapping("/sendOnewayMessage/{message}")
    public void sendOnewayMessage(@PathVariable("message")String data){
        Message<String> msg = MessageBuilder.withPayload(data).build();
        template.sendOneWay("topic-order:topic-order-flow",msg);
        System.out.println("sendOneway successfully");
    }
    
    // 延迟发送，syncSend的最后一个参数3，是第三等级，十秒
    @RequestMapping("/delaySend/{message}")
    public void delaySend(@PathVariable("message")String data){
        Message<String> msg = MessageBuilder.withPayload(data).build();
        template.syncSend("topic-order:topic-order-flow", msg, 3000, 3);
        System.out.println("sendOneway successfully");
    }
}

消费者

@Component
@RocketMQMessageListener(
        // 消费者名字
        consumerGroup = "consumer-group-zc",
        // 主题 和 producer一样
        topic = "topic-order",
        // tags 和 producer一样
        selectorExpression = "topic-order-flow",
        //消息消费模式：默认是CLUSTERING集群，还支持BROADCASTING广播
        messageModel = MessageModel.CLUSTERING
)
public class Consumer implements RocketMQListener<MessageExt> {
 
    @Override
    public void onMessage(MessageExt messageExt) {
        //这里拿到的消息 message.getBody 是byte[]格式。
        if(messageExt.getBody() == null || messageExt.getBody().length == 0){
            return;
        }
        //拿到消息：如果发送的消息是字符串，那么需要把byte[]转为字符串
        String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
        System.out.println(msg);
    }
}

效果

 Rocket插件效果