普通消息

RocketMQ提供三种方式来发送普通消息：可靠同步发送、可靠异步发送和单向发送。

可靠同步发送

同步发送是指消息发送放发出数据后，会在收到接收方响应之后才发下一个数据包的通讯方式。

这种方式应用场景非常广泛，列如重要通知邮件、报名短信通知、营销短信系统等。

//同步消息
   @Test
   public void testSyncSend() {
//参数一: topic， 如果想添加tag 可以使用"topic:tag"的写法
//参数二: 消息内容
      SendResult sendResult =
              rocketMQTemplate.syncSend("test-topic-1", "这是一条同步消息");                         
      System.out.println(sendResult);
   }

可靠异步发送

异步发送是指发送放发出数据后，不等接收方发回响应，接着发送下一个数据包的通讯方式。发送方通过回调接口接收服务器响应，并对响应结果进行处理。

异步发送一般用于链路耗时较长，对RT响应时间较为敏感的业务场景，列如用户视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

//异步消息
   @Test
   public void testAsyncSend() throws InterruptedException { 
        public void testSyncSendMsg() {
//参数一: topic, 如果想添加tag 可以使用"topic:tag"的写法
//参数二: 消息内容
//参数三: 回调函数, 处理返回结果
          rocketMQTemplate.asyncSend(
                "test-topic-1", 
                "这是一条异步消息",
          new SendCallback() {
             @Override
             public void onSuccess(SendResult sendResult) {             
                 System.out.println(sendResult);
              }
              @Override
              public void onException(Throwable throwable) {             
                 System.out.println(throwable);
              }
        });
//让线程不要终止
       Thread.sleep(30000000);
   }

单向发送

单向发送是指发送方只负责发送消息，不等待服务器回应没有回调函数触发，只发送请求不等待应答。

适合用于某些耗时非常短，但对可靠性要求并不高的场景，不如说日志收集。

//单向消息
      @Test
      public void testOneWay() {
         rocketMQTemplate.sendOneWay("test-topic-1", "这是一条单向消息");
      }

三种发送方式的对比

顺序消息

//同步顺序消息[异步顺序 单向顺序写法类似]
        public void testSyncSendOrderly() {
//第三个参数用于队列的选择
            rocketMQTemplate.syncSendOrderly(
                "test-topic-1", 
                "这是一条异步顺序消息",
                "xxxx");
        }

事物消息

RocketMQ提供了事务消息，通过事务消息就能达到分布式事务的最终一致。

两个概念

半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到 RocketMQ服务，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息为半事务消息。

消息回查：由于网络闪断、生产者应用重启等原因，导致的某条事务消息的二次确认丢失，

RocketMQ服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或者 Rollback）,该询问过程消息回查。

事务消息发送步骤

1.发送方将半事务消息发送至PocketMQ服务端。

2.RocketMQ服务端将消息持久化之后，向发送方返回Ack确认消息已经发送成功，此时消息为半事务消息。

3.发送方开始执行本地事务逻辑

4.发送方根据本地事务执行结果向服务端提交二次确认（Commit 或者 Rollback）,服务端收到Commit 状态则将半事务消息标记为可投递，订阅方最终将收到该消息。服务端说到R0llback状态则删除半事务消息，订阅方将不会接受该消息。

事务消息回查步骤

1.在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达服务端，经过固定时间后服务端将对该数据发起消息回查

2.发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。

3.发送方根据检查得到的本地事务的最终状态再次提交二次确认，服务端仍然按照步骤4对半事务消息进行操作。

//事物日志
@Entity(name = "shop_txlog")

@Data
public class TxLog {

        @Id
        private String txLogId;

private String content;

        private Date date;
}

@Service
public class OrderServiceImpl4 {
@Autowired

private OrderDao orderDao;

@Autowired
private TxLogDao txLogDao;

@Autowired
private RocketMQTemplate rocketMQTemplate;

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.setTxLogId(txId);

        txLog.setContent("事物测试");

        txLog.setDate(new Date());
txLogDao.save(txLog);
}
}

@RocketMQTransactionListener(txProducerGroup = "tx_producer_group")

public class OrderServiceImpl4Listener implements RocketMQLocalTransactionListener {

        @Autowired
private TxLogDao txLogDao;

@Autowired
private OrderServiceImpl4 orderServiceImpl4;

        //执行本地事物

@Override

public RocketMQLocalTransactionState executeLocalTransaction(

        Message msg, Object arg) {
try {
        //本地事物
orderServiceImpl4.createOrder((String) msg.getHeaders().get("txId"), (Order) arg);

            return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
return RocketMQLocalTransactionState.ROLLBACK;
}
}

//消息回查

    @Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        //查询日志记录
TxLog txLog = txLogDao.findById((String) msg.getHeaders().get("txId")).get();

if (txLog == null) {
return RocketMQLocalTransactionState.COMMIT;
} else {
return RocketMQLocalTransactionState.ROLLBACK;
}
}
}

消息消费要注意的细节

@RocketMQMessageListener(
consumerGroup = "shop",//消费者分组

topic = "order-topic",//要消费的主题
consumeMode = ConsumeMode.CONCURRENTLY, //消费模式:无序和有序 messageModel = MessageModel.CLUSTERING, //消息模式:广播和集群,默认是集群
)
public class SmsService implements RocketMQListener<Order> {

}