RocketMQ学习笔记
消息中间件应用场景
- 应用解耦
- 削峰填谷
- 数据分发
常见的消息中间件
- ActiveMQ：Apache出品，比较老的一个开源的消息中间件，以前在中小企业应用广泛
- Kafka：Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统。
- RabbitMQ：基于erlang语言开发的消息中间件，RabbitMQ最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。适用于对数据的一致性、稳定性和可靠性要求比较高的场景
- RocketMQ：高性能、低延时和高可靠性等特性
消息中间件对比
| Kafka | RocketMQ | RabbitMQ | |
|---|---|---|---|
| 定位 | 日志消息、监控数据 | 非日志的可靠消息传输 | 非日志的可靠消息传输 |
| 可用性 | 非常高、分布式、主从 | 非常高、分布式、主从 | 高、主从、采用镜像模式实现，数据量大时可能有性能问题 |
| 消息可靠性 | 异步刷盘，容易丢数据 | 同步刷盘、异步刷盘 | 同步刷盘 |
| 单机吞吐量 | 百万级 | 十万级 | 万级 |
| 堆积能力 | 非常好 | 非常好 | 一般 |
| 顺序消费 | 支持，一台broker宕机后，消息会乱序 | 支持，顺序消费场景下，消费失败时消费队列将会暂停 | 支持，如果一个消费失败，此消息的顺序会被打乱 |
| 定时消息 | 不支持 | 支持 | 支持 |
| 事务消息 | 不支持 | 支持 | 不支持 |
| 消息重试 | 不支持 | 支持 | 支持 |
| 死信队列 | 不支持 | 支持 | 支持 |
| 访问权限 | 无 | 无 | 类似数据库，配置用户名和密码 |
核心概念
- 消息生产者Producer：往RocketMQ发送消息的应用程序
- 消息消费者Consumer：从RocketMQ拉取消息，并根据消息执行业务的应用程序
- 代理服务器Broker：实际和消费生产者、消息消费者进行交互的程序，主要进行消息的存储、消息的推送，一般从性能考虑，会对消息进行集群
- 命名服务NameServer：代理服务器Broker在启动的时候注册信息到NameServer中，消息生产者和消息消费者启动的时候从NameServer拉取Broker的IP和端口，消息生产者应用程序和消息消费者的应用程序会和Broker建立长链接（基于Netty），发送消息和消费消息都是基于长链接的通道
- 主题Topic：划分不同类型的消息，比如订单消息发送到OrderTopic，会员消息发送到MemberTopic
- 消息队列MessageQueue：在Topic内部，专门进行消息的存储的地方，最底层存储的数据结构（默认一个Topic中有4个MessageQueue）
- 消息Message：在消息生产者给RocketMQ发送消息的时候，需要将传递的参数封装到Message对象中，通过网络传输到RocketMQ，消息会存储在MessageQueue中
- 标签Tag：可以在发送的时候给消息添加标签Tag，消费者可以通过标签Tag进行过滤
简单使用
引入jar包
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.5</version>
</dependency>
生产者
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;public class SimpleProducer {public static void main(String[] args) throws Exception{//定义消费的生产者对象DefaultMQProducer producer = new DefaultMQProducer("helloProducerGroup");//定义nameServer地址producer.setNamesrvAddr("127.0.0.1:9876");//自动连接（从nameServer拉取broker地址，并且建立连接）producer.start();//定义消息发送的目的地TopicString topic = "helloTopic";for(int i=0;i<10;i++){//定义消息Message message = new Message(topic,("helloTopic的消息=="+i).getBytes());//发送消息SendResult result = producer.send(message);//输出消息储存的结果System.out.println("消息存储的状态:"+result.getSendStatus());System.out.println("消息存储的消息ID:"+result.getMsgId());}//关闭连接producer.shutdown();}
}
消费者
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;public class SimpleConsumer {public static void main(String[] args) throws Exception {//定义消息的消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("helloConsumerGroup");//定义nameServer地址consumer.setNamesrvAddr("127.0.0.1:9876");//定义消费的主题String topic = "helloTopic";//监听该主题消息consumer.subscribe(topic,"*");//设置消息监听器，服务器把消息推送给我们，消费消息consumer.setMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {for(MessageExt messageExt:list){System.out.println("处理的线程："+Thread.currentThread()+"，消息内容："+new String(messageExt.getBody()));}//告诉消息中间件，消息处理的情况return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});consumer.start();}
}
)
