关于Java整合RocketMQ实现生产消费详解
作者:叫我二蛋 发布时间:2022-12-31 07:44:34
环境搭建
创建Maven项目。
pom.xml文件中引入RocketMQ依赖。
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
</dependency>
</dependencies>
生产者
普通消息
RocketMQ可用于以三种方式发送消息:同步、异步和单向传输。前两种消息类型是可靠的,因为无论它们是否成功发送都有响应。
同步发送
private final static String nameServer = "127.0.0.1:9876";
private final static String producerGroup = "my_group";
private final static String topic = "topic-test";
@Test
public void syncSend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 启动producer
producer.start();
// 创建一条消息,并指定topic、tag、body等信息,tag可以理解成标签,对消息进行再归类,RocketMQ可以在消费端对tag进行过滤
Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 利用producer进行发送,并同步等待发送结果
SendResult sendResult = producer.send(msg, 10000);
System.out.printf("%s%n", sendResult);
// 一旦producer不再使用,关闭producer
producer.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
异步发送
@Test
public void asyncSend() throws IOException {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 启动producer
producer.start();
// 创建一条消息,并指定topic、tag、body等信息,tag可以理解成标签,对消息进行再归类,RocketMQ可以在消费端对tag进行过滤
Message msg = new Message(topic, "tagB", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 异步发送消息, 发送结果通过callback返回给客户端
producer.send(msg, new SendCallback() {
public void onSuccess(SendResult sendResult) {
System.out.printf("OK %s %n",
sendResult.getMsgId());
}
public void onException(Throwable e) {
System.out.printf("Exception %s %n", e);
e.printStackTrace();
}
},10000);
} catch (Exception e) {
e.printStackTrace();
}
System.in.read();
}
单向传输
@Test
public void onewaySend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
producer.setSendMsgTimeout(10000);
// 启动producer
producer.start();
// 创建一条消息,并指定topic、tag、body等信息,tag可以理解成标签,对消息进行再归类,RocketMQ可以在消费端对tag进行过滤
Message msg = new Message(topic, "tagB", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 异步发送消息, 发送结果通过callback返回给客户端
producer.sendOneway(msg);
// 一旦producer不再使用,关闭producer
//producer.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
顺序消息
RocketMQ 通过生产者和服务端的协议保障单个生产者串行地发送消息,并按序存储和持久化。如需保证消息生产的顺序性,则必须满足以下条件:
单一生产者: 消息生产的顺序性仅支持单一生产者,不同生产者分布在不同的系统,即使设置相同的分区键,不同生产者之间产生的消息也无法判定其先后顺序。
串行发送:生产者客户端支持多线程安全访问,但如果生产者使用多线程并行发送,则不同线程间产生的消息将无法判定其先后顺序。
@Test
public void orderSend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
producer.setSendMsgTimeout(10000);
// 启动producer
producer.start();
String[] tags = new String[]{"TagA", "TagB", "TagC"};
for (int i = 0; i < 10; i++) {
int orderId = i % 10;
Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.printf("%s%n", sendResult);
}
} catch (Exception e) {
e.printStackTrace();
}
}
延迟消息
延迟消息发送是指消息发送到RocketMQ后,并不期望立马投递这条消息,而是延迟一定时间后才投递到Consumer进行消费。 使用 RocketMQ 的延时消息可以简化定时调度任务的开发逻辑,实现高性能、可扩展、高可靠的定时触发能力。 RocketMQ 一共支持18个等级的延迟投递,具体时间如下:
投递等级 | 延迟时间 | 投递等级 | 延迟时间 |
1 | 1s | 10 | 6min |
2 | 5s | 11 | 7min |
3 | 10s | 12 | 8min |
4 | 30s | 13 | 9min |
5 | 1min | 14 | 10min |
6 | 2min | 15 | 20min |
7 | 3min | 16 | 30min |
8 | 4min | 17 | 1h |
9 | 5min | 18 | 2h |
@Test
public void scheduledSend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 启动producer
producer.start();
// 创建一条消息,并指定topic、tag、body等信息,tag可以理解成标签,对消息进行再归类,RocketMQ可以在消费端对tag进行过滤
Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 消息延迟等级
msg.setDelayTimeLevel(2);
// 利用producer进行发送,并同步等待发送结果
SendResult sendResult = producer.send(msg, 10000);
System.out.printf("%s%n", sendResult);
// 一旦producer不再使用,关闭producer
producer.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
批量消息
在对吞吐率有一定要求的情况下,Apache RocketMQ可以将一些消息聚成一批以后进行发送,可以增加吞吐率,并减少API和网络调用次数。
@Test
public void batchSend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 启动producer
producer.start();
List<Message> messages = new ArrayList<Message>();
messages.add(new Message(topic, "Tag", "Order001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "Tag", "Order002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "Tag", "Order003", "Hello world 2".getBytes()));
// 利用producer进行发送,并同步等待发送结果
SendResult sendResult = producer.send(messages, 10000);
System.out.printf("%s%n", sendResult);
// 一旦producer不再使用,关闭producer
producer.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
事务消息
在一些对数据一致性有强需求的场景,可以用RocketMQ 事务消息来解决,从而保证上下游数据的一致性。
基于 RocketMQ 的分布式事务消息功能,在普通消息基础上,支持二阶段的提交能力。将二阶段提交和本地事务绑定,实现全局提交结果的一致性。
第一阶段会发送一个半事务消息,半事务消息是指暂不能投递的消息。 如果发送成功则执行本地事务,并根据本地事务执行成功与否,向 Broker 半事务消息状态(commit或者rollback)。 半事务消息只有 commit 状态才会真正向下游投递。 如果由于网络闪断、生产者应用重启等原因,导致某条事务消息的二次确认丢失,Broker 端会通过扫描发现某条消息长期处于“半事务消息”时,需要主动向消息生产者询问该消息的最终状态(Commit或是Rollback)。这样最终保证了本地事务执行成功,下游就能收到消息,本地事务执行失败,下游就收不到消息。
事务消息的详细交互流程如下图所示:
@Test
public void transactionSend() {
try {
// 事务消息的发送不再使用 DefaultMQProducer,而是使用 TransactionMQProducer 进行发送
TransactionMQProducer producer = new TransactionMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 事务回查的线程池
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
producer.setExecutorService(executorService);
producer.setTransactionListener(new TransactionListener() {
//半事务消息发送成功后,执行本地事务的方法
public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
System.out.printf("执行本地事务 %n");
/*
二次确认
LocalTransactionState.COMMIT_MESSAGE:提交事务,允许消费者消费该消息
LocalTransactionState.ROLLBACK_MESSAGE:回滚事务,消息将被丢弃不允许消费。
LocalTransactionState.UNKNOW:暂时无法判断状态,等待固定时间以后Broker端根据回查规则向生产者进行消息回查。
*/
return LocalTransactionState.UNKNOW;
}
// 二次确认消息没有收到,Broker端回查事务状态的方法,默认60s
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
System.out.printf("二次确认失败,broker事务回查 %n");
return LocalTransactionState.COMMIT_MESSAGE;
}
});
producer.setSendMsgTimeout(10000);
// 启动producer
producer.start();
Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 利用producer进行发送事务消息,并同步等待发送结果
SendResult sendResult = producer.sendMessageInTransaction(msg, null);
System.out.printf("%s%n", sendResult);
// 一旦producer不再使用,关闭producer
System.in.read();
} catch (Exception e) {
e.printStackTrace();
}
}
消费者
MQ的消费模式可以大致分为两种,一种是推Push,一种是拉Pull。
Push消费
Push是服务端主动推送消息给客户端,优点是及时性较好,但如果客户端没有做好流控,一旦服务端推送大量消息到客户端时,就会导致客户端消息堆积甚至崩溃。
private final static String nameServer = "127.0.0.1:9876";
private final static String consumerGroup = "my_group";
private final static String topic = "topic-test";
@Test
public void consumerPush() throws MQClientException, IOException {
// 初始化consumer,并设置consumer group name
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
// 设置NameServer地址
consumer.setNamesrvAddr(nameServer);
// 订阅一个或多个topic,并指定tag过滤条件,这里指定*表示接收所有tag的消息
consumer.subscribe(topic, "*");
//设置采用广播模式,广播模式下,消费组内的每一个消费者都会消费全量消息。
//consumer.setMessageModel(MessageModel.BROADCASTING);
//注册回调接口来处理从Broker中收到的消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
// 返回消息消费状态,ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 顺序消费
// consumer.registerMessageListener(new MessageListenerOrderly() {
// public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
// return ConsumeOrderlyStatus.SUCCESS;
// }
// });
// 启动Consumer
consumer.start();
System.out.printf("Consumer Started.%n");
System.in.read();
}
Pull 消费
Pull是客户端需要主动到服务端取数据,优点是客户端可以依据自己的消费能力进行消费,但拉取的频率也需要用户自己控制,拉取频繁容易造成服务端和客户端的压力,拉取间隔长又容易造成消费不及时。
@Test
public void consumerPull() {
try {
DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(consumerGroup);
consumer.setNamesrvAddr(nameServer);
//关闭自动提交
consumer.setAutoCommit(false);
consumer.subscribe(topic, "*");
consumer.setPullBatchSize(20);
consumer.start();
while (true) {
List<MessageExt> messageExts = consumer.poll();
System.out.printf("%s%n", messageExts);
consumer.commitSync();
}
} catch (Exception e) {
e.printStackTrace();
}
}
代码仓库
https://gitee.com/codeWBG/learn_rocketmq
来源:https://wangbinguang.blog.csdn.net/article/details/128452607
猜你喜欢
- 背景后台系统需要接入 企业微信登入,满足企业员工快速登入系统流程图简单代码说明自定义一套 springsecurity 认证逻辑主要就是 根
- 本文实例为大家分享了opencv实现轮廓高斯滤波平滑的具体代码,供大家参考,具体内容如下一个小测试的题目:在图像上点选,找到与点选处相邻的颜
- 前言提问:springboot项目,开发环境、测试环境和生产环境配置文件如何分开表示?答:多profile文件方式1、多环境配置(profi
- 对于数据的访问来说,肯定是在有缓存的情况下运行快一些。对于Hibernate这种与数据库结合紧密的框架来说,在调用数据的时候肯定会有缓存的出
- 一、定时器是什么定时器类似于我们生活中的闹钟,可以设定一个时间来提醒我们。而定时器是指定一个时间去执行一个任务,让程序去代替人工准时操作。标
- JNI简介JNI是Java Native Interface的缩写,它提供了若干的API实现了Java和其他语言的通信(主要是C&C
- 1、定时器推动整个计算机硬件的发展的核心关键性技术就是时钟。所以在企业开发中定时操作往往成为开发重点。而在JDK本身也支持这种定时调度的处理
- Oracle官网是这样介绍默认方法的,使用默认方法,可以达到往接口里面增加新的功能,而且保持与老版本代码兼容,也就是原来的实现类可以不需要被
- 题目要求思路一:模拟迭代依次判断每个节点是否合法:左子树判断是否>low,合法就向左下走,不合法往右下;右子树判断是否<high
- 上篇文章给大家介绍了在idea中将创建的java web项目部署到Tomcat中的过程图文详解,可以参考下,本文给大家继续介绍如何在IDEA
- strcpy函数详解如下1.函数介绍1.1.函数接口char * __cdecl strcpy(char * dst, const char
- 本文将带领大家使用基于JAX-RS REST风格的实现Jersey来上传文件到服务器制定的文件夹,如果是图片并读取显示出该图片。准备工作:准
- 2011年6月iBatis 更名为 MyBatis,从 iBatis 到 MyBatis,不只是名称上的变化,MyBatis 提供了更为强大
- jdk8之前 一、java.lang.Systemlong times = System.currentTimeMillis();
- 一. 前言最近我发现了一个事情,那就是在面试笔试中,好多公司都喜欢在String字符串上出问题,涉及到方方面面的知识,包括其中的一些常用方法
- 举个例子Map < String , Object > jsonMap = new HashMap< String , O
- 为什么要自定义缓存注解?Spring Cache本身提供@Cacheable、@CacheEvict、@CachePut等缓存注解,为什么还
- 前言为了应对在SpringBoot中的高并发及优化访问速度,我们一般会把页面上的数据查询出来,然后放到redis中进行缓存。减少数据库的压力
- 介绍 在开发过程中,我们有时候会遇到非接口调用而出发程序执行任务的一些场景,比如我们使用quartz定时框架通过配置文件来启动定时任务时,
- 前言日常开发中,我们可能会碰到需要进行防重放与操作幂等的业务,本文记录SpringBoot实现简单防重与幂等防重放,防止数据重复提交操作幂等