关于Java整合RocketMQ实现生产消费详解
作者：叫我二蛋
这篇文章主要介绍了关于Java整合RocketMQ实现生产消费详解,RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等,需要的朋友可以参考下
环境搭建
- 创建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
到此这篇关于关于Java整合RocketMQ实现生产消费详解的文章就介绍到这了,更多相关Java整合RocketMQ生产消费内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！