RocketMQ

一、什么是RocketMQ

主要特点：

应用场景：

二、RocketMQ的使用

添加依赖

配置RocketMQ

创建生产者服务

创建消费者监听器

编写测试Controller

启动Spring Boot应用

三、RocketMQ与Kafka的异同

相似点

不同点

一、什么是RocketMQ

RocketMQ 是一款分布式消息中间件，最初由阿里巴巴集团开发，并于2012年开源。它在2017年成为了Apache软件基金会的顶级项目。RocketMQ设计用于满足互联网分布式架构的需求，特别是在处理海量消息时提供高效、稳定的服务。

主要特点：

高吞吐量：RocketMQ能够支持每秒数万条至数十万条的消息发送和接收。
低延迟：优化了消息传输机制，确保即使在高负载下也能实现低延迟的消息传递。
可靠性：通过持久化存储、主从复制等机制保证消息不丢失，确保服务的高可用性。
可扩展性：设计上易于水平扩展，以适应业务增长的需求。
灵活性：提供了多种消息模型，包括点对点、发布/订阅模式，以及顺序消息、事务消息等功能。

应用场景：

异步通信：将原本同步的操作转换为异步处理，提高系统响应速度。
削峰填谷：在网络购物节等流量高峰期，使用消息队列可以有效缓解数据库的压力。
日志处理：作为大规模分布式日志收集和处理的解决方案。
事件驱动架构：构建基于事件驱动的应用程序和服务。

二、RocketMQ的使用

添加依赖

首先，在你的pom.xml文件中添加必要的依赖项：

<dependencies><!-- Spring Boot Starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- RocketMQ Spring Boot Starter --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.1</version> <!-- 确保选择适合的版本 --></dependency>
</dependencies>

配置RocketMQ

在application.properties或application.yml中配置RocketMQ的相关属性。这里以application.properties为例：

# RocketMQ NameServer地址
rocketmq.name-server=127.0.0.1:9876
# 生产者组名
rocketmq.producer.group=my-producer-group
# 消费者组名
rocketmq.consumer.group=my-consumer-group

创建生产者服务

创建一个服务类用于发送消息：

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class RocketMQProducerService {@Autowiredprivate RocketMQTemplate rocketMQTemplate;public void sendMessage(String topic, String message) {rocketMQTemplate.convertAndSend(topic, message);System.out.println("Sent Message: "message);}
}

创建消费者监听器

创建一个监听器类用于接收消息：

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "my-consumer-group")
public class RocketMQConsumerService implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.printf("Received Message: %s %n", message);}
}

编写测试Controller

为了方便测试，可以编写一个简单的Controller来调用生产者服务发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class TestController {@Autowiredprivate RocketMQProducerService producerService;@GetMapping("/send")public String sendMessage(@RequestParam String msg) {producerService.sendMessage("test-topic", msg);return "Message sent!";}
}

启动Spring Boot应用

确保RocketMQ的服务（NameServer和Broker）已经启动，并运行你的Spring Boot应用程序。你可以通过访问http://localhost:8080/send?msg=Hello_RocketMQ来发送一条消息，并观察控制台输出确认消息是否被成功消费。

三、RocketMQ与Kafka的异同

相似点

高吞吐量：两者都设计为能够处理大规模的数据流，并且都能够提供非常高的消息吞吐量。
持久化存储：两者都支持将消息持久化到磁盘，以确保即使发生故障也不会丢失数据。
扩展性：两者都可以通过增加节点来实现水平扩展，从而支持更高的负载。
顺序消息：都能保证特定条件下的消息顺序（如分区内的顺序）。

不同点

存储模型：
- Kafka使用基于日志结构的存储模型，消息按顺序追加到持久化日志文件中。
- RocketMQ采用更类似于传统数据库的设计，将消息存储在逻辑表中。
消息可靠性：
- Kafka的消息可靠性依赖于副本机制，但可能无法完全保证消息不丢失。
- RocketMQ通过重试等机制提高了消息保存的可靠性，并提供了同步刷盘功能，在单机可靠性上比Kafka更高。
消息模式：
- Kafka主要关注于高吞吐量场景，适用于大数据流处理。
- RocketMQ提供了更多的消息模型，如顺序消息、延时消息和事务消息，适合需要高可靠性和复杂消息模式的应用。
性能优化：
- Kafka在高吞吐量场景下表现优异，特别适合需要大量数据快速传输的情况。
- RocketMQ在高并发场景下有更好的表现，特别是在需要保证消息准确性的场景中。
API灵活性：
- Kafka拥有丰富的客户端库，支持多种编程语言，便于集成不同的系统。
- RocketMQ的API相对简单，但在某些高级特性上提供了更强的支持。
社区和支持：
- Kafka由Apache软件基金会维护，拥有广泛的社区支持和大量的第三方工具。
- RocketMQ同样也是Apache顶级项目，但其社区规模和生态系统相对较小。