RocketMQ学习笔记（一）

RocketMQ学习笔记

消息中间件应用场景

• 应用解耦
• 削峰填谷
• 数据分发

常见的消息中间件

• ActiveMQ：Apache出品，比较老的一个开源的消息中间件，以前在中小企业应用广泛
• Kafka：Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统。
• RabbitMQ：基于erlang语言开发的消息中间件，RabbitMQ最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。适用于对数据的一致性、稳定性和可靠性要求比较高的场景
• RocketMQ：高性能、低延时和高可靠性等特性

消息中间件对比

核心概念

• 消息生产者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();}
}