1.普通消息
普通消息是指消息队列RocketMQ版中无特性的消息,区别于有特性的定时和延时消息、顺序消息和事务消息。
同步发送
原理
同步发送是指消息发送方发出一条消息后,会在收到服务端同步响应之后才发下一条消息的通讯方式。sync
应用场景
此种方式应用场景非常广泛,例如重要通知邮件、报名短信通知、营销短信系统等。
@Autowired
private RocketMQTemplate rocketMQTemplate;
//同步消息
@Test
public void testSyncSend(){
//参数一: topic 添加tag 可以使用topic:tag
//参数二: 消息内容
SendResult sendResult = rocketMQTemplate.syncSend("test-topic-1:tag","这是一条同步消息");
System.out.println(sendResult);
}
异步发送
原理
异步发送是指发送方发出一条消息后,不等服务端返回响应,接着发送下一条消息的通讯方式。消息队列RocketMQ版的异步发送,需要您实现异步发送回调接口(SendCallback)。消息发送方在发送了一条消息后,不需要等待服务端响应即可发送第二条消息。发送方通过回调接口接收服务端响应,并处理响应结果。
应用场景
异步发送一般用于链路耗时较长,对响应时间较为敏感的业务场景,例如,您视频上传后通知启动转码服务,转码完成后通知推送转码结果等。
//异步消息
@Test
public void testAyncSend() throws InterruptedException{
//参数一: topic 添加tag 可以使用topic:tag
//参数二: 消息内容
//参数三: 回调结果,处理返回结果
rocketMQTemplate.asyncSend("test-topic-1:tag", "这是一条异步消息", new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
@Override
public void onException(Throwable throwable) {
System.out.println(throwable);
}
});
System.out.println("=========================");
Thread.sleep(30000000000L);
}
单向发送
原理
发送方只负责发送消息,不等待服务端返回响应且没有回调函数触发,即只发送请求不等待应答。此方式发送消息的过程耗时非常短,一般在微秒级别。
应用场景
适用于某些耗时非常短,但对可靠性要求并不高的场景,例如日志收集。
//单向消息
@Test
public void testOneWaySend(){
//参数一: topic 添加tag 可以使用topic:tag
//参数二: 消息内容
rocketMQTemplate.sendOneWay("test-topic-1:tag","这是一条单向消息");
}
2.顺序消息
顺序消息分为两类:
全局顺序:对于指定的一个Topic,所有消息按照严格的先入先出FIFO(First In First Out)的顺序进行发布和消费。
具体就是在后面添加一个Orderly
//单向顺序消息
@Test
public void sendOneWayOrderly(){
//参数一: topic 添加tag 可以使用topic:tag
//参数二: 消息内容
//参数三: 发送到哪个队列
rocketMQTemplate.sendOneWayOrderly("test-topic-1:tag","这是一条单向消息","xxx");
}
//同步顺序消息
@Test
public void testSyncSendOrderly(){
//参数一: topic 添加tag 可以使用topic:tag
//参数二: 消息内容
//参数三: 发送到哪个队列
rocketMQTemplate.syncSendOrderly("test-topic-1:tag","这是一条单向消息","xxx");
}
//异步顺序消息
@Test
public void testAyncSendOrderly() throws InterruptedException {
//参数一: topic 添加tag 可以使用topic:tag
//参数二: 消息内容
//参数三: 回调结果,处理返回结果
rocketMQTemplate.asyncSendOrderly("test-topic-1:tag", 10000, "这是一条异步消息", new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
@Override
public void onException(Throwable throwable) {
System.out.println(throwable);
}
});
System.out.println("=========================");
Thread.sleep(30000000000L);
}
# 3.事务消息
通过消息队列RocketMQ版事务消息,能达到分布式事务的最终一致。
交互流程
事务消息交互流程如下图所示。
首先在我们之前代码中添加相关Service类
![image.png](https://ucc.alicdn.com/pic/developer-ecology/8b95e6e12d5542a8a22bd482a7d16724.png)
@Service
@Slf4j
public class OrderServiceImplRocketMQ {
@Autowired
private OrderDao orderDao;
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private TxLogDao txLogDao;
public void createOrderBefore(Order order) {
String txId= UUID.randomUUID().toString();
rocketMQTemplate.sendMessageInTransaction("tx_producer_group", "tx_topic", MessageBuilder.withPayload(order).setHeader("txId",txId).build(), order);
}
@Transactional
public void createOrder(String txId,Order order) {
orderDao.save(order);
TxLog txLog = new TxLog();
txLog.setTxId(txId);
txLog.setDate(new Date());
//记录事务日志
txLogDao.save(txLog);
}
}
两个概念:
半事务消息:暂不能投递的消息,发送方已经成功地将消息发送到了RocketMQ服务端,但是服务端未收到生产者对该消息的二次确认,此时该消息被标记成"暂不能投递"状态,处于该种状态下的消息即半事务消息。
消息回查:由于网络闪断、生产者应用重启等原因,导致某条事务消息的二次确认丢失,RocketMQ服务端通过扫描发现某条消息长期处于"半事务消息"时,需要主动向消息生产者询问该消息的最终状态(Commit或是Rollback),该询问过程即消息回查。
编写Listenter
这样实现了事务。
后期会在这个项目上不断添加,喜欢的请点个start~
项目源码参考一下分支220311_xgc_rocketMQ
Gitee:https://gitee.com/coderxgc/springcloud-alibaba
GitHub:https://github.com/coderxgc/springcloud-alibaba