RocketMQ之Producer

Producer 介绍

消息发送的步骤

1
2
3
4
5
1) 设置 Producer 的 GroupName (Producer Group是一类 Producer 的集合)
2) 设置 InstanceName,当一个 JVM 需要启动多个 Producer 时,需要指定不同的 InstanceName 来区分,不显式设置时使用默认名称 "DEFAULT"
3) 设置发送失败重试次数,默认值是2次,可能会出现重复消息,因此需要消费端进行控制
4) 设置 NameServer 地址
5) 组装数据并发送

生产者核心参数

1
2
3
4
5
6
7
8
9
* producerGroup: 生产者组名
* createTopicKey: 创建 Topic,生产环境一般不直接从代码层面创建而是在控制台创建
* defaultTopicQueueNums: 每个 Topic 下的队列数量,默认数量是4
* sendMsgTimeout: 消息发送超时时间,单位ms
* compressMsgBodyOverHowmuch: 当消息大小超过指定字节就会开启压缩,默认字节为4096
* retryTimesWhenSendFailed: 同步模式下,消息发送失败重试次数,默认2次
* retryTimesWhenSendAsyncFailed: 异步模式下,消息发送失败重试次数,默认2次
* retryAnotherBrokerWhenNotStoreOK: 当 broker 接收失败时,是否切换另一个 broker,默认为 false
* maxMessageSize: 最大的消息容量限制,默认是4M

不同类型的生产者

生产者向消息队列中写入消息,根据不同的业务场景需要采用不同的写入策略,如同步发送、异步发送、延迟发送和发送事务消息等。

同步发送

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class Producer {

public static void main(String[] args) throws MQClientException, RemotingException,
InterruptedException, MQBrokerException {
// 创建生产者对象
DefaultMQProducer producer = new DefaultMQProducer("producerGroupName");
// 设置实例化名称
producer.setInstanceName("SyncProducer");
// 指定同步模式下,失败重试次数
producer.setRetryTimesWhenSendFailed(5);
// 设置服务器地址
producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER);
// 启动实例
producer.start();
// 实例化消息对象
Message message = new Message("topicTest", "tagA", "同步消息发送".getBytes());
// 同步发送消息
SendResult sendResult = producer.send(message);
System.out.printf("%s%n", sendResult);
// 关闭生产者
producer.shutdown();
}
}

异步发送

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public class Producer {

public static void main(String[] args) throws MQClientException, RemotingException,
InterruptedException, MQBrokerException {
// 创建生产者对象
DefaultMQProducer producer = new DefaultMQProducer("producerGroupName");
// 设置实例化名称
producer.setInstanceName("AsyncProducer");
// 指定异步模式下,失败重试次数
producer.setRetryTimesWhenSendAsyncFailed(5);
// 设置服务器地址
producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER);
// 启动实例
producer.start();
// 实例化消息对象
Message message = new Message("topicTest", "tagA", "异步消息发送".getBytes());
// 异步发送消息
producer.send(message, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("%s%n", sendResult);
// 关闭生产者
producer.shutdown();
}

@Override
public void onException(Throwable e) {
e.printStackTrace();
}
});
}
}

延迟发送

RocketMQ支持发送延迟消息,Broker在收到这类消息后,会延迟一段时间再处理,使消息在规定的一段时间后生效。

延迟消息的使用方法是在创建 Message 对象时,调用setDelayTimeLevel(int level)方法设置延迟时间。目前不支持自定义时间,只能使用预定义的时间长度,如setDelayTimeLevel(3)表示延迟10s。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class Producer {

public static void main(String[] args) throws MQClientException, RemotingException,
InterruptedException, MQBrokerException {
// 创建生产者对象
DefaultMQProducer producer = new DefaultMQProducer("producerGroupName");
// 设置实例化名称
producer.setInstanceName("SyncProducer");
// 指定同步模式下,失败重试次数
producer.setRetryTimesWhenSendFailed(5);
// 设置服务器地址
producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER);
// 启动实例
producer.start();
// 实例化消息对象
Message message = new Message("topicTest", "tagA", "延迟消息发送".getBytes());
// 设置延迟时间,时间长度为(1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h)
message.setDelayTimeLevel(3);
// 发送消息
SendResult sendResult = producer.send(message);
System.out.printf("%s%n", sendResult);
// 关闭实例
producer.shutdown();
}
}

自定义发送规则

一个 Topic 会有多个Message Queue, Producer 的默认配置会轮流向各个Message Queue发送消息。Consumer 在消费消息时,会根据负载均衡策略,消费被分配到的Message Queue。如果要把消息发送到指定的Message Queue,可以使用Message-QueueSelector

1
2
3
4
5
6
7
8
9
10
11
public class MyMessageQueueSelector implements MessageQueueSelector {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object key) {\
// 自定义选择 Message Queue 规则
int id = Integer.parseInt(key.toString());
int idMainIndex = id / 100;
int size = mqs.size();
int index = idMainIndex % size;
return mqs.get(index);
}
}

在发送消息的时候,将MessageQueueSelector对象作为参数即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class Producer {

public static void main(String[] args) throws MQClientException, RemotingException,
InterruptedException, MQBrokerException {
// 创建生产者对象
DefaultMQProducer producer = new DefaultMQProducer("producerGroupName");
// 设置实例化名称
producer.setInstanceName("SyncProducer");
// 指定同步模式下,失败重试次数
producer.setRetryTimesWhenSendFailed(5);
// 设置服务器地址
producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER);
// 启动实例
producer.start();
// 实例化消息对象
Message message = new Message("topicTest", "tagA", "自定义消息发送".getBytes());
// 发送消息
SendResult sendResult = producer.send(message, new MyMessageQueueSelector(), 1000);
System.out.printf("%s%n", sendResult);
// 关闭实例
producer.shutdown();
}
}

事务消息

RocketMQ事务消息流程

当某几件事需要同时成功或失败的时候,就需要使用到事务功能,如银行转账:A 银行的某账户要转一万元到 B 银行的某账户

1
2
1) 从 A 账户扣除一万元
2) 对 B 账户增加一万元

这两个操作需要同时成功或同时失败,RocketMQ 采用两阶段提交的方式实现事务消息,TransactionMqRroducer处理流程如下:

RocketMQ_Transaction

1
2
3
4
5
6
7
8
9
1) 发送方向 RocketMQ 发送 "待确认" 消息
2) RocketMQ 将收到的 "待确认" 消息持久化成功后,向发送方回复消息已经发送成功,此时第一阶段消息发送完成
3) 发送发开始执行本地事件逻辑
4) 发送方根据本地事件执行结果向 RocketMQ 发送二次确认(Commit 或 Rollback) 消息:
* 接收到 commit 消息,将把第一阶段消息标记为可投递,订阅方将会收到该消息
* 接收到 rollback 消息,将删除第一阶段消息,订阅方不会接受到该消息
5) 如果出现异常情况,步骤4 提交的二次确认最终未到达 RocketMQ,服务器将经过固定时间段后将对 "待确认" 消息发起回查请求
6) 发送方收到消息回传请求后(如果第一阶段发送的 Producer 不能工作时,将会回传给同一个 ProducerGroup 的其他 Producer),通过对检查对应消息的本地事件执行结果返回 Commit 或 Rollback 状态
7) RocketMQ 收到回查请求后,按照步骤4) 流程继续处理

RocketMQ 通过以下几个类来支持用户实现事务消息:

  • TransactionMQProducer

DefaultMQProducer用户类似,通过它启动事务消息,相比DefaultMQProducer需要多设置本地事务处理函数和回查状态函数

  • TransactionListener

提供本地执行方法和回查方法,返回LocalTransactionState状态标识:

1
2
3
* LocalTransactionState.COMMIT_MESSAGE: 提交
* LocalTransactionState.ROLLBACK_MESSAGE: 回滚
* LocalTransactionState.UNKNOW: 未知,需要回查

RocketMQ事务消息实现

实现 TransactionListener 接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
public class MyTransactionListener implements TransactionListener {

private AtomicInteger transactionIndex = new AtomicInteger(0);
private AtomicInteger checkTimes = new AtomicInteger(0);

private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

// 执行本地事务
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
String msgKey = msg.getKeys();
System.out.println("start execute local transaction " + msgKey);
LocalTransactionState state;
if (msgKey.contains("1")) {
// 第一条消息让他通过
state = LocalTransactionState.COMMIT_MESSAGE;
} else if (msgKey.contains("2")) {
// 第二条消息模拟异常,明确回复回滚操作
state = LocalTransactionState.ROLLBACK_MESSAGE;
} else {
// 第三条消息无响应,让它调用回查事务方法
state = LocalTransactionState.UNKNOW;
// 给剩下3条消息,放1,2,3三种状态
localTrans.put(msgKey, transactionIndex.incrementAndGet());
}
System.out.println("executeLocalTransaction:" + msg.getKeys() + ",execute state:" + state + ",current time:" + System.currentTimeMillis());
return state;
}

// 检查本地事务结果
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
String msgKey = msg.getKeys();
System.out.println("start check local transaction " + msgKey);
Integer state = localTrans.get(msgKey);
switch (state) {
case 1:
System.out.println("check result unknown 回查次数" + checkTimes.incrementAndGet());
return LocalTransactionState.UNKNOW;
case 2:
System.out.println("check result commit message, 回查次数" + checkTimes.incrementAndGet());
return LocalTransactionState.COMMIT_MESSAGE;
case 3:
System.out.println("check result rollback message, 回查次数" + checkTimes.incrementAndGet());
return LocalTransactionState.ROLLBACK_MESSAGE;
default:
return LocalTransactionState.COMMIT_MESSAGE;
}
}
}

实现事务消息生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class Producer {

public static void main(String[] args) throws MQClientException, RemotingException,
InterruptedException, MQBrokerException {
// 创建事务生产者对象
TransactionMQProducer producer = new TransactionMQProducer("producerGroupName");
// 设置实例化名称
producer.setInstanceName("SyncProducer");
// 指定同步模式下,失败重试次数
producer.setRetryTimesWhenSendFailed(5);
// 设置事务监听器
producer.setTransactionListener(new MyTransactionListener());
// 设置服务器地址
producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER);

// 启动实例
producer.start();
for (int i = 0; i < 5; i++) {
// 实例化消息对象
Message message = new Message("topicTest", "tagA", "msg-" + i, ("事务消息发送" + ":" + i).getBytes());
// 发送消息
SendResult sendResult = producer.sendMessageInTransaction(message, i);
System.out.printf("%s%n", sendResult);
}

// 关闭实例
// producer.shutdown();
}

}

消息返回状态

获取消息发送返回状态对象

1
2
3
4
// 获取发送消息结果
SendResult sendResult = producer.send(message);
// 从结果中获取状态对象
SendStatus sendStatus = sendResult.getSendStatus();

状态值

1
2
3
4
5
6
7
8
9
10
public enum SendStatus {
// 表示发送成功
SEND_OK,
// 表示没有在指定时间内完成刷盘(需要 Broker 的刷盘策略被设置成SYNC_FLUSH)
FLUSH_DISK_TIMEOUT,
// 表示在主备模式下,并且 Broker 被设置成SYNC_MASTER,没有在指定时间内完成主从同步
FLUSH_SLAVE_TIMEOUT,
// 表示在主备模式下,并且 Broker 被设置成SYNC_MASTER,没有找到被配置成 Slave 的 Broker
SLAVE_NOT_AVAILABLE,
}

注:当返回状态不是SEND_OK时,都需要有补偿机制

Powered by AppBlog.CN     浙ICP备14037229号

Copyright © 2012 - 2021 APP开发技术博客 All Rights Reserved.

访客数 : | 访问量 :