rocketmq的使用

MQ是中大型项目优先考虑的消息队列中间件，其优点是解耦、异步、消峰，目前主流有RabbitMQ、RocketMQ、ActiveMQ、Kafka等

之前用的是rabbitmq，rocketmq也是第一次使用

首先，来到官方Apache RocketMQ开发者指南 https://github.com/apache/rocketmq/tree/master/docs/cn

了解下其概念，看看样例 其表面工作机制与传统mq还是一样的 无非是生产者生产消息，消费者消费消息

对于单机应用 在pom中引入依赖即

    org.apache.rocketmq
    rocketmq-client
    4.3.0

集成到springboot项目 引入OpenService即可，不了解openService的可以阿里云搜索下消息服务MNS

对于单体应用集成 参考下官方样例：https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

集成OpenService 需要添加消息服务依赖至pom中

    com.aliyun.mns
    aliyun-sdk-mns
    1.1.9

配置文件

rocketmq:
    accessKey: ${rocketmq.accessKey}
    secretKey: ${rocketmq.secretKey}
    nameSrvAddr: ${rocketmq.nameSrvAddr}
    # 【生产者配置】初始化普通消息类型生产者
    initProducer: true
    # 【生产者配置】初始化顺序消息类型生产者
    initOrderProducer: true
    # 【消费者配置】初始化普通消息类型消费者
    initConsumer: true
    # 【消费者配置】初始化顺序消息类型消费者
    initOrderConsumer: false
    # 【生产者配置】发送超时时间，单位毫秒. 不配置采用默认5000
    sendTimeOut: ${rocketmq.sendTimeOut}
    #【消费者配置】groupId
    groupId: ${rocketmq.groupId}
    #【消费者配置】消费者线程数,不设置rocketmq默认为20.
    consumeThreadNums: ${rocketmq.consumeThreadNums}
    topic:
      status: ${rocketmq.topic.pay.status}
      Msg: ${rocketmq.topic.refund.Msg}

项目中使用的是顺序消息，因此使用OrderProducerBean实例进行发送消息

orderProducerBean.send(msg);

对应的，对于顺序消息的消费 需要Consumer实现MessageOrderListener监听器来实现接受以及处理消息

@Slf4j
@Component
public class Consumer implements MessageOrderListener {
	 @Override
	 public OrderAction consume(Message message, ConsumeOrderContext context) {
	 	try{
	 		log.info("【接受到消息】Message：{}，key:{},tag:{}", message, key, tag);
			//处理消息
			return OrderAction.Success;
		}catch{
			log.error(errorMsg);
			//消息消费失败 重试
			return OrderAction.Suspend;
		}
	 }

上面返回的OrderAction对象 是消费结果 MNS会根据返回的数据判断是否重试

对于消息如何保证幂等，可以参考阿里云给的解释：https://help.aliyun.com/document_detail/44397.html

欢迎大家共同交流