RocketMQ 常用消息类型

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</dependency>

rocketmq:
  name-server: 172.16.250.129:9876
  producer:
    group: myGroup

public void sendMsg() throws Exception {

    Message message = new Message(
            // 普通消息所属的Topic
            "Topic-Normal",
            // Message Tag可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在消息队列 RocketMQ 的服务器过滤。
            "TagA",
            // Message Body可以是任何二进制形式的数据。
            "Hello MQ".getBytes()
    );
    rocketMQTemplate.getProducer().send( message );
    // 等同于上面的方式(常用)
    //rocketMQTemplate.convertAndSend("Topic-Normal:TagA","Hello MQ".getBytes());
}

public void sendAsyncMsg() {
    Map<String , Object> map = new HashMap<>();
    map.put( "name" , "zs" );
    map.put( "age" , 20);

    rocketMQTemplate.asyncSend( "Topic-Normal", map , new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            // 消息发送成功。
            log.info( "async send success" );
        }

        @Override
        public void onException(Throwable throwable) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
            log.info( "async send fail" );
        }
    } );
}

public void sendOrderlyMsg() {
    //根据指定的hashKey按顺序发送
    for (int i = 0; i < 1000; i++) {
        String orderId = "biz_" + i % 10;
        // 分区顺序消息中区分不同分区的关键字段，Sharding Key与普通消息的key是完全不同的概念。
        // 全局顺序消息，该字段可以设置为任意非空字符串。
        String shardingKey = String.valueOf(orderId);
        try {
            SendResult sendResult = rocketMQTemplate.syncSendOrderly( "Topic-Order", "send order msg".getBytes(), shardingKey );
            // 发送消息，只要不抛异常就是成功。
            if (sendResult != null) {
                System.out.println(new Date() + " Send mq message success . msgId is:" + sendResult.getMsgId());
            }
        }
        catch (Exception e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
            System.out.println(new Date() + " Send mq message failed");
            e.printStackTrace();
        }
    }
}

public void sendDelayMsg() {
    rocketMQTemplate.syncSend( "Topic-Delay",
            MessageBuilder.withPayload( "Hello MQ".getBytes() ).build(),
            3000,
            //设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
            //messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
            3 );
}

/**
 * 事务消息
 */
public void sendTransactionMsg() {
    TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(
            "Topic-Tx:TagA",
            MessageBuilder.withPayload( "Hello MQ transaction===".getBytes() ).build(),
            null );
    SendStatus sendStatus = transactionSendResult.getSendStatus();
    LocalTransactionState localTransactionState = transactionSendResult.getLocalTransactionState();
    System.out.println( new Date() + " Send mq message status "+ sendStatus +" ,  localTransactionState "+ localTransactionState );
}

@Component
@RocketMQTransactionListener
public class TxProducerListener implements RocketMQLocalTransactionListener {

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        System.out.println("TX message listener execute local transaction");
        RocketMQLocalTransactionState result;
        try {
            // 业务代码（ 例如下订单 ）
            result = RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            System.out.println("execute local transaction error");
            result = RocketMQLocalTransactionState.UNKNOWN;
        }
        return result;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 检查本地事务（ 例如检查下订单是否成功 ）
        System.out.println("TX message listener check local transaction");
        RocketMQLocalTransactionState result;
        try {
            //业务代码（ 根据检查结果，决定是COMMIT或ROLLBACK ）
            result = RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            // 异常就回滚
            System.out.println("check local transaction error");
            result = RocketMQLocalTransactionState.ROLLBACK;
        }
        return result;
    }

}

@Component
@Slf4j
@RocketMQMessageListener(topic = "Topic-Tx",consumerGroup = "consumer-tx-group")
public class TxConsumerListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        log.info("Receive message：{}" , message);
    }

}