文章目录
创建项目
pom文件
新建 maven 项目或 module,添加 rocketmq-client 依赖。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.tedu</groupId>
<artifactId>rocketmq-api</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.7.1</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-store</artifactId>
<version>4.7.1</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
同步消息
同步消息发送要保证强一致性,发到master的消息向slave复制后,才会向生产者发送反馈信息。
这种可靠性同步地发送方式使用的比较广泛,比如:重要的消息通知,短信通知。
生产者
package m1;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
//新建生产者实例
DefaultMQProducer p = new DefaultMQProducer("producer1");
//设置 name server 地址
p.setNamesrvAddr("192.168.64.141:9876");
//启动生产者 连接服务器
p.start();
//消息数据封装到 Message 对象
//发送
while(true){
System.out.println("输入消息:");
String s = new Scanner(System.in).nextLine();
/**
* Topic --- 一级分类
* Tag --- 二级分类(可选)
*/
//发送的目的地、发送给谁、发送的消息
Message msg = new Message("Topic1", "TagA", s.getBytes());//Topic1是自己在服务器上创建的
SendResult r = p.send(msg);
System.out.println("发送的消息:"+r);
}
}
}
消费者
消费者的要点:
1.push 和 pull
消费者有两种模式:push 和 pull。
push 模式由服务器主动向消费者发送消息;pull 模式由消费者主动向服务器请求消息。
在消费者处理能力有限时,为了减轻消费者的压力,可以采用pull模式。多数情况下都采用 pull 模式。
2.NameServer
消费者需要向 NameServer 询问 Topic 的路由信息。
3.Topic
从指定的Topic接收消息。Topic相当于是一级分类。
4.Tag
Topic 相当于是一级分类,Tag 相当于是2级分类。
- 多个 Tag 可以这样写:
TagA || TagB || TagC
- 不指定 Tag,或者说接收所有的 Tag,可以写星号:
*
package m1;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws MQClientException {
//新建消费者实例
DefaultMQPushConsumer c = new DefaultMQPushConsumer("Consumer1");
//设置 name server
c.setNamesrvAddr("192.168.64.141:9876");
//订阅消息(从哪订阅消息)
/**
* 标签设置:
* TagA
* TagB || TagC || TagD
* *
*/
c.subscribe("Topic1", "TagA");//从topic1订阅标签A的消息
//消息监听器
//Concurrently监听器会启动多个线程,可以并行的处理多条消息
c.setMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt ext : msgs) {//快捷键:msgs.for
String s = new String(ext.getBody());
System.out.println("收到:"+s);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
});
//启动
c.start();
}
}
异步消息
master 收到消息后立即向生产者进行反馈。之后再以异步方式向 slave 复制消息。
异步消息通常用在对响应时间敏感的业务场景,即发送端不能容忍长时间地等待Broker的响应。
生产者
package demo2;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
/*
异步发送消息
一条消息送出后, 不必暂停等待服务器针对这条消息的反馈, 而是可以立即发送后续消息.
使用监听器, 以异步的方式接收服务器的反馈
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo2");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
p.setRetryTimesWhenSendAsyncFailed(0);
String topic = "Topic2";
String tag = "TagA";
String key = "Key-demo2";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, tag, key, s.getBytes());
p.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("\n\n消息发送成功 : "+sendResult);
}
@Override
public void onException(Throwable throwable) {
System.out.println("\n\n消息发送失败");
}
});
System.out.println("--------------------消息已送出-----------------------");
}
}
}
}
消费者
package demo2;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/*
与 demo1.Consumer 完全相同
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo2");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic2", "TagA");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
单向消息
这种方式主要用在不特别关心发送结果的场景,例如日志发送。
生产者
package demo3;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
/*
单向消息
消息发出后, 服务器不会返回结果
*/
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo3");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
String topic = "Topic3";
String tag = "TagA";
while (true) {
System.out.print("输入消息,用逗号分隔多条消息: ");
String[] a = new Scanner(System.in).nextLine().split(",");
for (String s : a) {
Message msg = new Message(topic, tag, s.getBytes());
p.sendOneway(msg);
}
System.out.println("--------------------消息已送出-----------------------");
}
}
}
消费者
package demo3;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/*
与 demo1.Consumer 完全相同
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo2");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic3", "TagA");
c.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
System.out.println(new String(msg.getBody()) + " - " + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
顺序消息
上图演示了 Rocketmq 顺序消息的基本原理:
- 同一组有序的消息序列,会被发送到同一个队列,按照 FIFO 的方式进行处理
- 一个队列只允许一个消费者线程接收消息,这样就保证消息按顺序被接收
下面以订单为例:
一个订单的顺序流程是:创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中。消费时,从同一个队列接收同一个订单的消息。
生产者
package demo4;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.List;
import java.util.Scanner;
/*
以下消息, 相同id的消息按顺序发送到同一个队列,
消费时也从同一个队列按顺序消费
topic
======================= queue1
======================= queue2
111,消息1 111,消息2 111,消息3 ------->======================= queue3
======================= queue4
222,消息1 222,消息2 222,消息3 ------->======================= queue5
======================= queue6
333,消息1 333,消息2 333,消息3 ------->======================= queue7
======================= queue8
......
*/
public class Producer {
static String[] msgs = {
"15103111039,创建",
"15103111065,创建",
"15103111039,付款",
"15103117235,创建",
"15103111065,付款",
"15103117235,付款",
"15103111065,完成",
"15103111039,推送",
"15103117235,完成",
"15103111039,完成"
};
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer p = new DefaultMQProducer("producer-demo4");
p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
p.start();
String topic = "Topic4";
String tag = "TagA";
for (String s : msgs) {
System.out.println("按回车发送此消息: "+s);
new Scanner(System.in).nextLine();
Message msg = new Message(topic, tag, s.getBytes());
String[] a = s.split(",");
long orderId = Long.parseLong(a[0]);
/*
MessageQueueSelector用来选择发送的队列,
这里用订单的id对队列数量取余来计算队列索引
send(msg, queueSelector, obj)
第三个参数会传递到queueSelector, 作为它的第三个参数
*/
SendResult r = p.send(msg, new MessageQueueSelector() {
/*
三个参数的含义:
queueList: 当前Topic中所有队列的列表
message: 消息
o: send()方法传入的orderId
*/
@Override
public MessageQueue select(List<MessageQueue> queueList, Message message, Object o) {
Long orderId = (Long) o;
//订单id对队列数量取余, 相同订单id得到相同的队列索引
long index = orderId % queueList.size();
System.out.println("消息已发送到: "+queueList.get((int) index));
return queueList.get((int) index);
}
}, orderId);
System.out.println(r+"\n\n");
}
}
}
消费者
package demo4;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo4");
c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
c.subscribe("Topic4", "*");
c.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
String t = Thread.currentThread().getName();
for (MessageExt msg : list) {
System.out.println(t+" - "+ msg.getQueueId() + " - " +new String(msg.getBody()));
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
c.start();
System.out.println("开始消费数据");
}
}
延时消息
消息发送到 Rocketmq 服务器后, 延迟一定时间再向消费者进行投递。
延时消息的使用场景:
比如电商里,提交了一个订单就可以发送一个延时消息,1h后去检查这个订单的状态,如果还是未付款就取消订单释放库存。
生产者发送消息时,对消息进行延时设置:
msg.setDelayTimeLevel(3);
其中3
代表级别而不是一个具体的时间值,级别和延时时长对应关系是在MessageStoreConfig
类种进行定义的:
this.messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
对应关系表:
级别 | 延时时长 |
---|---|
1 | 1s |
2 | 5s |
3 | 10s |
4 | 30s |
5 | 1m |
6 | 2m |
7 | 3m |
8 | 4m |
9 | 5m |
10 | 6m |
11 | 7m |
12 | 8m |
13 | 9m |
14 | 10m |
15 | 20m |
16 | 30m |
17 | 1h |
18 | 2h |
生产者
package m1;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.Scanner;
public class Producer {
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
//新建生产者实例
DefaultMQProducer p = new DefaultMQProducer("producer1");
//设置 name server 地址
p.setNamesrvAddr("192.168.64.141:9876");
//启动生产者 连接服务器
p.start();
//消息数据封装到 Message 对象
//发送
while(true){
System.out.println("输入消息:");
String s = new Scanner(System.in).nextLine();
/**
* Topic --- 一级分类
* Tag --- 二级分类(可选)
*/
//发送的目的地、发送给谁、发送的消息
Message msg = new Message("Topic1", "TagA", s.getBytes());//Topic1是自己在服务器上创建的
if(Math.random() < 0.5){
msg.setDelayTimeLevel(3);//代表10秒
System.out.println("这条消息延时10秒");
}
SendResult r = p.send(msg);
System.out.println("发送的消息:"+r);
}
}
}
消费者
package m1;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
/**
* @Authod yuhe
* Create 2021-10-29-16:40
*/
public class Consumer {
public static void main(String[] args) throws MQClientException {
//新建消费者实例
DefaultMQPushConsumer c = new DefaultMQPushConsumer("Consumer1");
//设置 name server
c.setNamesrvAddr("192.168.64.141:9876");
//订阅消息(从哪订阅消息)
/**
* 标签设置:
* TagA
* TagB || TagC || TagD
* *
*/
c.subscribe("Topic1", "TagA");//从topic1订阅标签A的消息
//消息监听器
//Concurrently监听器会启动多个线程,可以并行的处理多条消息
c.setMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt ext : msgs) {//快捷键:msgs.for
String s = new String(ext.getBody());
System.out.println("收到:"+s);
}
//return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
/**
* 在消息处理失败时,可以告诉服务器,稍后重新发送消息,重新消费。
* 如果多次处理失败,最多会重试 18 次(18个延时级别),重试的时间间隔会越来越长
*/
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
});
//启动
c.start();
}
}
批量消息
批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic,相同的waitStoreMsgOK,而且不能是延时消息。此外,这一批消息的总大小不应超过4MB。
生产者
消费者
消息过滤
Tag 过滤
Tag 可以满足大多数消息过滤的需求。使用 Tag 过滤非常简单,例如:
consumer.subscribe("Topic1", "TagA || TagB || TagC");
对自定义属性过滤
生产者可以在消息中添加自定义的属性:
msg.putUserProperty("prop1", "1");
msg.putUserProperty("prop2", "2");
消费者接收数据时,可以根据属性来过滤消息:
consumer.subscribe("Topic7", MessageSelector.bySql("prop1=1 or prop2=2"));
可以看到,自定义属性的过滤语法是 Sql 语法,RocketMQ只定义了一些基本语法来支持这个特性,支持的 Sql 过滤语法如下:
- 数值比较,比如:>,>=,<,<=,BETWEEN,=;
- 字符比较,比如:=,<>,IN;
- IS NULL 或者 IS NOT NULL;
- 逻辑符号 AND,OR,NOT;
生产者
消费者
事务消息
RocketMQ 提供了可靠性消息,也叫事务消息。下面分析一下其原理。
事务消息的原理
下面来看 RocketMQ 的事务消息是如何来发送“可靠消息”的,只需要以下三步:
- 发送半消息(半消息不会发送给消费者)
- 执行本地事务
- 提交消息
完成事务消息发送后,消费者就可以以正常的方式来消费数据。
RocketMQ 的自动重发机制在绝大多数情况下,都可以保证消息被正确消费。
假如消息最终消费失败了,还可以由人工处理进行托底。
上面分析的是正常情况下的执行流程。下面再来看两种错误情况:
- 事务执行失败时回滚消息
- 服务器无法得知消息状态时,需要主动回查消息状态
回滚:
消息回查:
生产者
消费者