RocketMQ学习

1.基本概念

RocketMQ是阿里巴巴团队使用java语言开发的一款分布式消息中间件，是一款低延迟，高可用，拥有海量消息堆积能力和灵活拓展性的消息队列。

rocketmq的官网：http://rocketmq.apache.org

gitee仓库：https://gitee.com/apache/rocketmq?_from=gitee_search

Apache RocketMQ开发者指南：https://www.itmuch.com/books/rocketmq/

RocketMQ主要由 Producer、Broker、Consumer 三部分组成。

1.1、Producer

消息生产者，负责生产消息，一般由业务系统负责生产消息。
发送消息到Broker服务器。
发送方式：
- 同步发送、
- 异步发送、
- 顺序发送、
- 单向发送。
同步和异步方式均需要Broker返回确认信息，单向发送不需要。

1.2、Consumer

消息消费者，负责消费消息，一般是后台系统负责异步消费。
从Broker服务器拉取消息。
消费形式：
- 拉取式消费、
- 推动式消费。

1.3、Broker

Broker 在实际部署过程中对应一台服务器，
一个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。
Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。

Topic: 消息的集合，一个主题有多条消息，每条消息只能属于某一个主题。

Broker Server：代理服务器，负责存储消息、转发消息。

Name Server：名字服务，生产者或消费者通过名字服务查找主题相应的Broker IP列表。

Message：消息，生产和消费数据的最小单位，每条消息必须属于一个主题。

Tag：标签，用于同一主题下区分不同类型的消息，可以根据不同业务目的在同一主题下设置不同标签。

2、架构设计

1 技术架构

RocketMQ架构上主要分为四部分:

Producer：支持分布式集群方式部署。
Consumer：支持分布式集群方式部署。
NameServer：
- 是一个非常简单的Topic路由注册中心，支持Broker的动态注册与发现。
- Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；
- 路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。
BrokerServer：
- 主要负责消息的存储、投递和查询以及服务高可用保证。

2 工作流程

启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

3、项目使用

官方例子：https://gitee.com/apache/rocketmq/blob/develop/docs/cn/RocketMQ_Example.md

3.1、Rocketmq下载

rocketmq的官网

1、下载zip 文件解压缩到本地磁盘中。例：D:\\rocketmq

2、启动服务，进入bin，先执行，mqnamesrv.cmd，再启动broker，通过命令行来指定端口。

mqbroker.cmd -n localhost:9876

3.2、监控平台下载

https://github.com/apache/rocketmq-dashboard

3.3、集成springboot项目

1、添加依赖

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.4</version>
</dependency>

2、添加配置信息

rocketmq:
  name-server: http://127.0.0.1:9876 #rocketmq服务地址
  producer:
    group: base_group_syncMsg
    send-message-timeout: 5000
    retry-times-when-send-failed: 2
    max-message-size: 4194304

4、测试例子
生产者

@RestController
@RequestMapping(\"/mq\")
public class ProducerController {

    @Autowired
    private RocketMQTemplate mqTemplate;

    @RequestMapping(\"/send\")
    public String testSend(String msg) {
        try {
            mqTemplate.convertAndSend(\"TopicTest\", msg);
            return \"success\";
        } catch (Exception e) {
            e.printStackTrace();
            return \"fail\";
        }
    }
}

消费者

@Component
@RocketMQMessageListener(topic = \"TopicTest\", consumerGroup = \"consumerGroupTest\")
@Slf4j
public class ConsumeController implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        log.init(\"------- Consumer: {}\", message);
    }
}

4、幂等性

RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，务必要在业务层面进行去重处理。

4.1、什么是幂等性

对一个事情进行操作，这个操作可能会执行成百上千次，但是操作结果都是相同的，这就是幂等性。

4.2、常用解决方案

业界主流的幂等性有两种操作：

唯一 ID + 指纹码机制，利用数据库主键去重
- 缺点，每次都要查询数据库
利用redis的原子性去实现
- 缺点：可能业务执行失败，但redis标识成功。

来源：https://www.cnblogs.com/galenblog/p/16277332.html
本站部分图文来源于网络，如有侵权请联系删除。

spring boot RocketMQ 集成