中国产品设计网_企查查app下载安装_企业推广公司_百度客服电话人工服务热线电话

目录

一、RocketMQ简介

        1.1、核心概念

        1.2、应用场景

        1.3、架构设计

2、RocketMQ Server安装

3、RocketMQ可视化控制台安装与使用

4、SpringBoot整合RocketMQ实现消息发送和接收 

        4.1、添加maven依赖

        4.2、yaml配置

        4.3、生产者

        4.4、消费者

        4.5、接口

        4.6、接口测试

---

一、RocketMQ简介

RocketMQ是由阿里捐赠给Apache的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。经历 了淘宝双十一的洗礼。RocketMQ既可为分布式应用系统提供异步解耦和削峰填谷的能力，同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。

官方文档：https://rocketmq.apache.org/docs/quick-start/

github中文主页：https://github.com/apache/rocketmq/tree/master/docs/cn

        1.1、核心概念

1. Topic：消息主题，一级消息类型，生产者向其发送消息。Message：生产者向Topic发送并最终传送给消费者的数据消息的载体。
2. 消息属性：生产者可以为消息定义的属性，包含Message Key和Tag。
3. Message Key：消息的业务标识，由消息生产者（Producer）设置，唯一标识某个业务逻辑。
4. Message ID：消息的全局唯一标识，由消息队列RocketMQ系统自动生成，唯一标识某条消息。
5. Tag：消息标签，二级消息类型，用来进一步区分某个Topic下的消息分类
6. Producer：也称为消息发布者，负责生产并发送消息至Topic。
7. Consumer：也称为消息订阅者，负责从Topic接收并消费消息。
8. 分区：即Topic Partition，物理上的概念。每个Topic包含一个或多个分区。
9. 消费位点：每个Topic会有多个分区，每个分区会统计当前消息的总条数，这个称为最大位点MaxOffset；分区的起始位置对应的位置叫做起始位点MinOffset。
10. Group：一类生产者或消费者，这类生产者或消费者通常生产或消费同一类消息，且消息发布或订 阅的逻辑一致。
11. Group ID：Group的标识。
12. 队列：单个Topic下会由一到多个队列来存储消息。
13. Exactly-Once投递语义：Exactly-Once投递语义是指发送到消息系统的消息只能被Consumer处理且仅处理一次，即使Producer重试消息发送导致某消息重复投递，该消息在Consumer也只被消费 一次。
14. 集群消费：一个Group ID所标识的所有Consumer平均分摊消费消息。例如某个Topic有9条消息，一个Group ID有3个Consumer实例，那么在集群消费模式下每个实例平均分摊，只消费其中的3条 消息。
15. 广播消费：一个Group ID所标识的所有Consumer都会各自消费某条消息一次。例如某个Topic有9条消息，一个Group ID有3个Consumer实例，那么在广播消费模式下每个实例都会各自消费9条消息。
16. 定时消息：Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息。
17. 延时消息：Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到Consumer进行消费，该消息即延时消息。
18. 事务消息：RocketMQ提供类似X/Open XA的分布事务功能，通过消息队列RocketMQ的事务消息能达到分布式事务的最终一致。
19. 顺序消息：RocketMQ提供的一种按照顺序进行发布和消费的消息类型，分为全局顺序消息和分区顺序消息。
20. 全局顺序消息：对于指定的一个Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。
21. 分区顺序消息：对于指定的一个Topic，所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段，和普通消息的Message Key是完全不同的概念。
22. 消息堆积：Producer已经将消息发送到消息队列RocketMQ的服务端，但由于Consumer消费能力有限，未能在短时间内将所有消息正确消费掉，此时在消息队列RocketMQ的服务端保存着未被消费的消息，该状态即消息堆积。
23. 消息过滤：Consumer可以根据消息标签（Tag）对消息进行过滤，确保Consumer最终只接收被过滤后的消息类型。消息过滤在消息队列RocketMQ的服务端完成。
24. 消息轨迹：在一条消息从Producer发出到Consumer消费处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹，您能清晰定位消息从Producer发出，经由消息队列RocketMQ服务端，投递给Consumer的完整链路，方便定位排查问题。
25. 重置消费位点：以时间轴为坐标，在消息持久化存储的时间范围内（默认3天），重新设置Consumer对已订阅的Topic的消费进度，设置完成后Consumer将接收设定时间点之后由Producer发送到消息队列RocketMQ服务端的消息。
26. 死信队列：死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，消息队列RocketMQ会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明Consumer在正常情况下无法正确地消费该消息。此时，消息队列RocketMQ不会立刻将消息丢弃，而是将这条消息发送到该Consumer对应的特殊队列中。消息队RocketMQ将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。

解压后的目录：

 

 

 

修改后保存，在项目目录下进入命令行，执行： 

D:\soft\rocketmqClient\rocketmq-externals-master\rocketmq-console

mvn clean package -Dmaven.test.skip=true

java -jar rocketmq-console-ng-2.0.0.jar

启动成功后，浏览器输入：http://localhost:19876/ 

说明一切OK；

4、SpringBoot整合RocketMQ实现消息发送和接收 

        4.1、添加maven依赖

<!--rocketmq依赖-->
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version>
</dependency>

        4.2、yaml配置

rocketmq:name-server: localhost:9876producer:# 发送同一类消息的设置为同一个group，保证唯一group: producer-group# 发送消息超时时间，默认3000sendMessageTimeout: 5000# 发送消息失败重试次数，默认2retryTimesWhenSendFailed: 2# 异步消息重试此处，默认2retryTimesWhenSendAsyncFailed: 2# 消息最大长度，默认1024 * 1024 * 4(默认4M)maxMessageSize: 4096# 压缩消息阈值，默认4k(1024 * 4)compressMessageBodyThreshold: 4096# 是否在内部发送失败时重试另一个broker，默认falseretryNextServer: falseconsumer:group: consumer-group # 消费者组名

        4.3、生产者

@Service
public class RocketMQProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;public void sendMessage(String topic, String message) {rocketMQTemplate.convertAndSend(topic, message);System.out.println("Message sent: " + message);}
}

        4.4、消费者

@Service
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "consumer-group")
public class RocketMQConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received message: " + message);}
}

        4.5、接口

    @GetMapping("/send")public String sendMessage(@RequestParam String topic, @RequestParam String message) {rocketMQProducer.sendMessage(topic, message);return "Message sent: " + message;}

        4.6、接口测试

 

  至此测试完成！