RocketMQ简单入门实例

墨蓝 2021-09-26 13:50 465阅读 0赞

### 消息队列优势 ###

专业&验证

1.  MQ是消息领域业内专业的消息中间件，多次在国内外获奖
2.  经过双11阿里交易、商品、营销等核心链路真实场景验证
3.  阿里集团内部1000+核心应用使用，每天流转几千亿条消息，稳定可靠
4.  3种消息发送方式，消息场景全覆盖：可靠同步、可靠异步、oneway方式

技术体系

1.  产品历史超过7年，消息保证不丢，技术体系丰富成熟
2.  阿里内部产品名：MetaQ、Notify
3.  开源社区产品名：RocketMQ，无技术绑定风险
4.  应用灵活，无任何强制绑定其他产品

独立部署

1.  支持专有云独立输出，支持物理机和虚拟机，最小部署仅几台机器
2.  专有云配套运维系统，方便运维人员实时监控系统状态
3.  专有云配套mqadmin命令集和管理类open API，方便集成及统一运维
4.  支持混合云架构，包括VPC用户

高可靠

1.  一份消息多份落盘存储，经过严格断电测试，消息依然保证不丢失
2.  支持消息轨迹，消息从生产到消费轨迹，可清晰排查
3.  海量消息堆积，单个Topic可堆积100亿+条消息，防止系统高流量崩溃
4.  默认情况下消息落盘保留3天

高性能

1.  同一网络内，消息传输RT在10毫秒之内，性能测试下，网卡可被打满
2.  公有云默认单Topic 发送消息为每秒5000条，最高可申请扩展至10W以上, 支持大量消息并发发送，超过5万个队列，性能依然卓越
3.  支持消息海量堆积，单Topic可堆积100+亿条消息
4.  单条消息默认最大支持256K，北京为4M

多协议

1.  HTTP接入（简单、免费）：HTTP公网接入支持跨网络调用，无短连接池，不会导致FULL GC，无CPU 100%占用风险
2.  MQTT接入（物联、免费）：支持主动推送模，多级Topic模型支持一次触达1000万+终端, 可广泛应用于物联网\\社交\\娱乐互动场景，支持SSL加密、web socket、flash
3.  TCP接入（专业） ：区别于HTTP简单的接入方式，同时提供更为专业、可靠、稳定的TCP协议的SDK接入

1.Producer代码如下:

package com.rocketmq;  
import com.alibaba.rocketmq.client.exception.MQClientException;  
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;  
import com.alibaba.rocketmq.client.producer.SendResult;  
import com.alibaba.rocketmq.common.message.Message;  
  
  
public class Producer \{  
public static void main(String\[\] args) throws MQClientException,  
InterruptedException \{  
/\*\*  
\* 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例  
\* 注意：ProducerGroupName需要由应用来保证唯一  
\* ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，  
\* 因为服务器会回查这个Group下的任意一个Producer  
\*/  
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");  
producer.setNamesrvAddr("127.0.0.1:80");  
/\*\*  
\* Producer对象在使用之前必须要调用start初始化，初始化一次即可 注意：切记不可以在每次发送消息时，都调用start方法  
\*/  
producer.start();  
  
  
/\*\*  
\* 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。  
\* 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态  
\* 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，  
\* 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。  
\*/  
for (int i = 0; i < 10; i++)  
try \{  
\{  
Message msg = new Message("Topic1",// topic  
"TagA",// tag  
"001",// key  
("Send Msg:Hello MetaQ1").getBytes());// body  
SendResult sendResult = producer.send(msg);  
System.out.println(sendResult);  
\}  
  
  
\{  
Message msg = new Message("Topic2",// topic  
"TagB",// tag  
"002",// key  
("Send Msg:Hello MetaQ2").getBytes());// body  
SendResult sendResult = producer.send(msg);  
System.out.println(sendResult);  
\}  
  
  
\{  
Message msg = new Message("Topic3",// topic  
"TagC",// tag  
"003",// key  
("Send Msg:Hello MetaQ3").getBytes());// body  
SendResult sendResult = producer.send(msg);  
System.out.println(sendResult);  
\}  
\} catch (Exception e) \{  
e.printStackTrace();  
\}  
  
  
/\*\*  
\* 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己  
\* 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法  
\*/  
producer.shutdown();  
\}  
\}

2.Customer代码如下:

package com.rocketmq;  
import java.util.List;  
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;  
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
import com.alibaba.rocketmq.client.exception.MQClientException;  
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;  
import com.alibaba.rocketmq.common.message.MessageExt;  
  
  
public class Consumer \{  
  
/\*\*  
\* 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。  
\* 但是实际PushConsumer内部是使用长轮询Pull方式从Broker拉消息，然后再回调用户Listener方法  
\*/  
public static void main(String\[\] args) throws InterruptedException,  
MQClientException \{  
/\*\*  
\* 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例  
\* 注意：ConsumerGroupName需要由应用来保证唯一  
\*/  
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID\_001");  
// consumer.setNamesrvAddr("10.10.0.102:9876");  
consumer.setNamesrvAddr("127.0.0.1:80");  
  
  
/\*\*  
\* 订阅指定topic下tags分别等于TagA或TagC或TagD  
\*/  
consumer.subscribe("Topic1", "TagA || TagC || TagD");  
/\*\*  
\* 订阅指定topic下所有消息<br>  
\* 注意：一个consumer对象可以订阅多个topic  
\*/  
consumer.subscribe("Topic2", "\*");  
consumer.subscribe("Topic3", "\*");  
  
  
/\*\*  
\* 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费 如果非第一次启动，那么按照上次消费的位置继续消费  
\*/  
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME\_FROM\_LAST\_OFFSET);  
consumer.registerMessageListener(new MessageListenerConcurrently() \{  
/\*\*  
\* 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息  
\*/  
public ConsumeConcurrentlyStatus consumeMessage(  
List<MessageExt> msgs, ConsumeConcurrentlyContext context) \{  
System.out.println(Thread.currentThread().getName()  
\+ " Receive New Messages: " + msgs);  
  
  
MessageExt msg = msgs.get(0);  
if (msg.getTopic().equals("Topic1")) \{  
if (null != msg.getTags()) \{  
// 执行Topic1的消费逻辑  
if (msg.getTags().equals("TagA")) \{  
// 执行TagA的消费  
System.out.println("TagA开始。");  
\} else if (msg.getTags().equals("TagC")) \{  
System.out.println("TagC开始。");  
// 执行TagC的消费  
\} else if (msg.getTags().equals("TagD")) \{  
// 执行TagD的消费  
System.out.println("TagD开始。");  
\}  
\}  
\} else if (msg.getTopic().equals("Topic2")) \{  
// 执行Topic2的消费逻辑  
System.out.println("Topic2");  
\}  
return ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;  
\}  
\});  
/\*\*  
\* Consumer对象在使用之前必须要调用start初始化，初始化一次即可  
\*/  
consumer.start();  
System.out.println("Consumer Started.");  
\}  
\}