RocketMq消息队列应用实战-1

public class OrderedProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        MQProducer producer = new DefaultMQProducer("example_group_name");
        //Launch the instance.
        producer.start();
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 100; i++) {
            int orderId = i % 10;
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Integer id = (Integer) arg;
                int index = id % mqs.size();
                return mqs.get(index);
            }
            }, orderId); // orderId就是arg，根据orderId取模得到存储的具体队列，实现同一个订单id的消息都放到同一队列里

            System.out.printf("%s%n", sendResult);
        }
        //server shutdown
        producer.shutdown();
    }
}

rocketMQTemplate.sendOneWayOrderly("settlement-test:test",jmap,jmap.get("orderId").toString());

public class SelectMessageQueueByHash implements MessageQueueSelector {
    public SelectMessageQueueByHash() {
    }

    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        int value = arg.hashCode() % mqs.size();
        if (value < 0) {
            value = Math.abs(value);
        }
        return (MessageQueue)mqs.get(value);
    }
}

public class OrderedConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("TopicTest", "TagA || TagC || TagD");

        consumer.registerMessageListener(new MessageListenerOrderly() {

            AtomicLong consumeTimes = new AtomicLong(0);
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeOrderlyContext context) {
                context.setAutoCommit(false);
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    return ConsumeOrderlyStatus.SUCCESS;
                } else if ((this.consumeTimes.get() % 3) == 0) {
                    return ConsumeOrderlyStatus.ROLLBACK;
                } else if ((this.consumeTimes.get() % 4) == 0) {
                    return ConsumeOrderlyStatus.COMMIT;
                } else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;

            }
        });

        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

@Service
public class OrderProducer{
  @Autowired
  private RocketMQTemplate rocketMQTemplate;

  @Value("${rocketmq.producer.send-message-timeout}")
  private Integer messageTimeOut;

  /**
     * 发送普通消息
     */
  public void sendMsg(String msgBody){
    rocketMQTemplate.syncSend("queue_test_topic",MessageBuilder.withPayload(msgBody).build());
  }

  /**
     * 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
     */
  public void sendAsyncMsg(String msgBody){
   rocketMQTemplate.asyncSend("queue_test_topic",MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
      @Override
      public void onSuccess(SendResult sendResult) {
        // 处理消息发送成功逻辑
      }
      @Override
      public void onException(Throwable e) {
        // 处理消息发送异常逻辑
      }
    });
  }

  /**
     * 发送延时消息<br/>
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h<br/>
     */
  public void sendDelayMsg(String msgBody, Integer delayLevel){    rocketMQTemplate.syncSend("queue_test_topic",MessageBuilder.withPayload(msgBody).build(),messageTimeOut,delayLevel);
  }

  /**
     * 发送带tag的消息,直接在topic后面加上":tag",看源码RocketMQUtil会把第一入参destination转为topic与tag
     */
  public void sendTagMsg(String msgBody){    rocketMQTemplate.syncSend("queue_test_topic:tag1",MessageBuilder.withPayload(msgBody).build());
  }

  public void testSend(Map jmap){
    rocketMQTemplate.convertAndSend("settlement-test:test",jmap); // 底层也是调用syncSend方法
  }
}

rocketMQTemplate.syncSend("RetryPushTodo:Tag1", MessageBuilder.withPayload(pushEntity).setHeader(RocketMQHeaders.KEYS,firstEntity.getModelId()).build());

@RunWith(SpringRunner.class)
@SpringBootTest
public class OrderSender {
    @Autowired
    OrderProducer orderProducer;

    @Test
    public void testProducer(){
        for(int i =0;i<10;i++){
            Map<String,Object> jmap= new HashMap<>(3);
            jmap.put("name","jacob"+i);
            jmap.put("age",i);
            orderProducer.testSend(jmap);
        }
        System.out.println("发送完毕");
    }
}

// selectorExpression 就是发送端的tag,默认是*,即接受所有tag的消息
@Service
@RocketMQMessageListener(topic = "settlement-test",selectorExpression = "test",consumerGroup = "settlement-ar-test")
public class OrderReceiver implements RocketMQListener<Map> {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

  // 只要没有抛异常，就会消费成功，请不要捕获异常，否则不会重新拉取消息进行重新消费
    @Override
    public void onMessage(Map message) {
        System.out.println("收到消息");
        System.out.println(message);
    }
}

    public void subscribe(String topic, String subExpression) throws MQClientException {
        try {
            SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(), topic, subExpression);
            this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
            if (this.mQClientFactory != null) {
                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
            }

        } catch (Exception var4) {
            throw new MQClientException("subscription exception", var4);
        }
    }

public interface RocketMQLocalTransactionListener {
  /*
   - 发送prepare消息成功此方法被回调，该方法用于执行本地事务
   - @param msg 回传的消息，利用transactionId即可获取到该消息的唯一Id
   - @param arg 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
   - @return 返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
     */
  RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg);
  
  /*
   - @param msg 通过获取transactionId来判断这条消息的本地事务执行状态
   - @return 返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
     */
  RocketMQLocalTransactionState checkLocalTransaction(Message msg);
}

TransactionMQProducer producer = new TransactionMQProducer("ProducerGroup");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
//设置TransactionListener实现
producer.setTransactionListener(transactionListener）；
//发送事务消息，
SendResult sendResult = producer.sendMessageInTransaction(msg, null);

@Mapper
@Component
public interface AccountInfoDao {
    @Update("update account_info set account_balance=account_balance+#{amount} where account_no=#{accountNo}")
    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

    @Select("select count(1) from de_duplication where tx_no = #{txNo}")
    int isExistTx(String txNo);

    @Insert("insert into de_duplication values(#{txNo},now());")
    int addTx(String txNo);
}

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {
	@Resource
	private RocketMQTemplate rocketMQTemplate;

	@Autowired
	private AccountInfoDao accountInfoDao;

	/**
	 * 更新帐号余额-发送消息
	 * producer向MQ Server发送消息
	 * @param accountChangeEvent
	 */
	@Override
	public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
		//构建消息体
		JSONObject jsonObject = new JSONObject();
		jsonObject.put("accountChange",accountChangeEvent);
		Message<String> message = MessageBuilder.withPayload(jsonObject.toJSONString()).build();
		TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg_bank1", "topic_txmsg:tag1", message, null);

		log.info("send transcation message body={},result={}",message.getPayload(),sendResult.getSendStatus());
	}

	/**
	 * 更新帐号余额-本地事务
	 * producer发送消息完成后接收到MQ Server的回应即开始执行本地事务
	 * @param accountChangeEvent
	 */
	@Transactional
	@Override
	public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
		log.info("开始更新本地事务，事务号：{}",accountChangeEvent.getTxNo());
		accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount() * -1);
		//为幂等作准备
		accountInfoDao.addTx(accountChangeEvent.getTxNo());
		if(accountChangeEvent.getAmount() == 2){
			throw new RuntimeException("bank1更新本地事务时抛出异常");
		}
		log.info("结束更新本地事务，事务号：{}",accountChangeEvent.getTxNo());
	}
}

@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {
    @Autowired
    AccountInfoService accountInfoService;
    @Autowired
    AccountInfoDao accountInfoDao;

    //消息发送成功回调此方法，此方法执行本地事务
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg){
        //解析消息内容
        try {
            String jsonString = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(jsonString);
            AccountChangeEvent accountChangeEvent = JSONObject.parseObject(jsonObject.getString("accountChange"), AccountChangeEvent.class);
            //扣除金额
            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("executeLocalTransaction 事务执行失败",e);
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    //此方法检查事务执行状态
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        RocketMQLocalTransactionState state;
        final JSONObject jsonObject = JSON.parseObject(new String((byte[]) message.getPayload()));
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(jsonObject.getString("accountChange"),AccountChangeEvent.class);

        //事务id
        String txNo = accountChangeEvent.getTxNo();
        int isexistTx = accountInfoDao.isExistTx(txNo);
        log.info("回查事务，事务号: {} 结果: {}", accountChangeEvent.getTxNo(),isexistTx);
        if(isexistTx>0){
            state=  RocketMQLocalTransactionState.COMMIT;
        }else{
            state=  RocketMQLocalTransactionState.UNKNOWN;
        }
        return state;
    }
}

@RestController
@Slf4j
public class AccountInfoController {
    @Autowired
    private AccountInfoService accountInfoService;

    @GetMapping(value = "/transfer")
    public String transfer(@RequestParam("accountNo")String accountNo,@RequestParam("amount") Double amount){
        String tx_no = UUID.randomUUID().toString();
        AccountChangeEvent accountChangeEvent = new AccountChangeEvent(accountNo,amount,tx_no);
				// 更新帐号余额-发送消息
        accountInfoService.sendUpdateAccountBalance(accountChangeEvent);
        return "转账成功";
    }
}

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {
  @Autowired
  AccountInfoDao accountInfoDao;
  
  /**
     * 消费消息，更新本地事务，添加金额
     * @param accountChangeEvent
     */
  @Override
  @Transactional
  public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {
    log.info("bank2更新本地账号，账号：{},金额：{}",accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());

    //幂等校验
    int existTx = accountInfoDao.isExistTx(accountChangeEvent.getTxNo());
    if(existTx<=0){
 		  //执行更新    	   
      accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
      //添加事务记录
      accountInfoDao.addTx(accountChangeEvent.getTxNo());
      log.info("更新本地事务执行成功，本次事务号: {}", accountChangeEvent.getTxNo());
    }else{
      log.info("更新本地事务执行失败，本次事务号: {}", accountChangeEvent.getTxNo());
    }
  }
}

@Component
@RocketMQMessageListener(topic = "topic_txmsg",selectorExpression="tag1",consumerGroup = "consumer_txmsg_group_bank2")
@Slf4j
public class TxmsgConsumer implements RocketMQListener<String> {
    @Autowired
    AccountInfoService accountInfoService;

    @Override
    public void onMessage(String s) {
        log.info("开始消费消息:{}",s);
        //解析消息为对象
        final JSONObject jsonObject = JSON.parseObject(s);
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(jsonObject.getString("accountChange"),AccountChangeEvent.class);
        //调用service增加账号金额
        accountChangeEvent.setAccountNo("2");
        accountInfoService.addAccountInfoBalance(accountChangeEvent);
    }
}

/**
 * @Author xiejw17
 * @Date 2021/9/22 20:27
 * 分销转自营，生成折让、红冲折让的销售单的结果订阅消费
 */
@Service
@RocketMQMessageListener(topic="TOPIC-MCSP-SMC-SETTLE-REBACK",selectorExpression="TAG-SETTLEBILL-RCC",consumerGroup="rcc-producer-group")
public class DistributeDiscountReversalConsumer implements RocketMQListener<MessageExt> {
    @Autowired
    BillProcessingLogService billProcessingLogService;

    @Autowired
    DistributeToOwnDetailMapper distributeToOwnDetailMapper;

    // 只要没有抛异常，就会消费成功，请不要捕获异常，否则不会重新拉取消息进行重新消费
    @Override
    public void onMessage(MessageExt msg) {
        String json = new String(msg.getBody());
        //log.info("生成折让消息体{}",json);
        SettleBillCallBackDTO dto = JSON.parseObject(json,SettleBillCallBackDTO.class);
        Map<String,Object> param = new HashMap<>(5);
        String id = dto.getSrcBillNum().substring(2);
        param.put("id", Long.parseLong(id));
        UpdateWrapper<BillProcessingLog> updateWrapper = new UpdateWrapper();
        if("Y".equals(dto.getProcessFlag())){
            // 1、成功，更新折让单单号，状态，原单号=前缀+id，
            param.put("orderNum",dto.getSettleBillNum()); // 单号
            param.put("state","11"); // 11-已审核
            if(dto.getSrcBillNum().startsWith("DI")){
                // 生成折让
                distributeToOwnDetailMapper.updateDiscount(param);
            }else{
                // 红冲折让
                distributeToOwnDetailMapper.updateReversalDiscount(param);
            }
            updateWrapper.set(BillProcessingLog.STATUS, BillProcessStatusEnum.COMPLETE.getStatus())
                    .set(BillProcessingLog.DETAIL,"success");
        }else{
            // 失败
            updateWrapper.set(BillProcessingLog.STATUS,BillProcessStatusEnum.EXCEPTION.getStatus())
                    .set(BillProcessingLog.DETAIL,dto.getResultMsg().length()>400?dto.getResultMsg().substring(0,400):dto.getResultMsg());
        }
        // 2、更新对账记录日志
        updateWrapper.eq("id",id);
        updateWrapper.set("updated_by","rocketMq");
        updateWrapper.set("update_time", LocalDateTime.now());
        billProcessingLogService.update(updateWrapper);
    }
}

@Bean(destroyMethod = "destroy")
@ConditionalOnBean(DefaultMQProducer.class)
@ConditionalOnMissingBean(name = RocketMQConfigUtils.ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer,
                                         ObjectMapper rocketMQMessageObjectMapper,
                                         RocketMQMessageConverter rocketMQMessageConverter) {
  RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
  rocketMQTemplate.setProducer(mqProducer);
  rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
  rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
  return rocketMQTemplate;
}

@Bean
@ConditionalOnBean(name = RocketMQConfigUtils.ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
@ConditionalOnMissingBean(TransactionHandlerRegistry.class)
public TransactionHandlerRegistry transactionHandlerRegistry(
  @Qualifier(RocketMQConfigUtils.ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME) RocketMQTemplate template) {
  return new TransactionHandlerRegistry(template);
}

@Bean(name = RocketMQConfigUtils.ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME)
@ConditionalOnBean(TransactionHandlerRegistry.class)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public static RocketMQTransactionAnnotationProcessor transactionAnnotationProcessor(
  TransactionHandlerRegistry transactionHandlerRegistry) {
  return new RocketMQTransactionAnnotationProcessor(transactionHandlerRegistry);
}

private void registerContainer(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

  // 需要实现接口RocketMQListener，否则抛异常
        if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
        }
  // 利用反射获取注解信息
        RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
  // 从注解你获取消费组、topic名称
        String consumerGroup = this.environment.resolvePlaceholders(annotation.consumerGroup());
        String topic = this.environment.resolvePlaceholders(annotation.topic());
  // 判断是否不订阅注解中的消费组和topic，是则不创建消费者
        boolean listenerEnabled =
            (boolean)rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
                .getOrDefault(topic, true);

        if (!listenerEnabled) {
            log.debug(
                "Consumer Listener (group:{},topic:{}) is not enabled by configuration, will ignore initialization.",
                consumerGroup, topic);
            return;
        }
  // 校验
        validate(annotation);
		// 监听者容器bean名称
        String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
            counter.incrementAndGet());
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext)applicationContext;
// 创建消费者实例，注册进ioc容器
        genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
            () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
        DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
            DefaultRocketMQListenerContainer.class);
 // 启动消费者，开始监听
        if (!container.isRunning()) {
            try {
                container.start();
            } catch (Exception e) {
                log.error("Started container failed. {}", container, e);
                throw new RuntimeException(e);
            }
        }

        log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
    }

public static final class Consumer {
  /**
         * listener configuration container
         * the pattern is like this:
         * group1.topic1 = false
         * group2.topic2 = true
         * group3.topic3 = false
         */
  private Map<String, Map<String, Boolean>> listeners = new HashMap<>();

  public Map<String, Map<String, Boolean>> getListeners() {
    return listeners;
  }

  public void setListeners(Map<String, Map<String, Boolean>> listeners) {
    this.listeners = listeners;
  }
}

 genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
            () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));

    private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,
        RocketMQMessageListener annotation) {
        DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
        
        container.setRocketMQMessageListener(annotation);
        
        String nameServer = environment.resolvePlaceholders(annotation.nameServer());
        nameServer = StringUtils.isEmpty(nameServer) ? rocketMQProperties.getNameServer() : nameServer;
        String accessChannel = environment.resolvePlaceholders(annotation.accessChannel());
        container.setNameServer(nameServer);
        if (!StringUtils.isEmpty(accessChannel)) {
            container.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }
        container.setTopic(environment.resolvePlaceholders(annotation.topic()));
        String tags = environment.resolvePlaceholders(annotation.selectorExpression());
        if (!StringUtils.isEmpty(tags)) {
            container.setSelectorExpression(tags);
        }
        container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));
        container.setRocketMQMessageListener(annotation);
        container.setRocketMQListener((RocketMQListener)bean); // 具体的业务消费者对象
        container.setObjectMapper(objectMapper);
        container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        container.setName(name);  // REVIEW ME, use the same clientId or multiple?

        return container;
    }

if (!container.isRunning()) {
  try {
    container.start();
  } catch (Exception e) {
    log.error("Started container failed. {}", container, e);
    throw new RuntimeException(e);
  }
}

DefaultMQPushConsumer consumer;

public interface InitializingBean {

	/**
	 * Invoked by the containing {@code BeanFactory} after it has set all bean properties
	 * and satisfied {@link BeanFactoryAware}, {@code ApplicationContextAware} etc.
	 * <p>This method allows the bean instance to perform validation of its overall
	 * configuration and final initialization when all bean properties have been set.
	 * @throws Exception in the event of misconfiguration (such as failure to set an
	 * essential property) or if initialization fails for any other reason
	 */
	void afterPropertiesSet() throws Exception;
}

public enum MessageModel {
    BROADCASTING("BROADCASTING"), // 广播消息，一个消费组内的每一个消费者都会消费topic的所有队列，使用场景刷新本地缓存，一个队列对应每一个消费者
    CLUSTERING("CLUSTERING");  // 集群消息，一个消费组内的每一个消费者只消费topic的部分队列，一个队列只对应一个消费者，

    private final String modeCN;

    private MessageModel(String modeCN) {
        this.modeCN = modeCN;
    }

    public String getModeCN() {
        return this.modeCN;
    }
}

switch (consumeMode) {
  case ORDERLY:
    consumer.setMessageListener(new DefaultMessageListenerOrderly());
    break;
  case CONCURRENTLY:
    consumer.setMessageListener(new DefaultMessageListenerConcurrently());
    break;
  default:
    throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
}

public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {

  @SuppressWarnings("unchecked")
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    for (MessageExt messageExt : msgs) {
      log.debug("received msg: {}", messageExt);
      try {
        long now = System.currentTimeMillis();
        // 把消息转换后，调用自定义的业务消费类，消费消息
        rocketMQListener.onMessage(doConvertMessage(messageExt));
        long costTime = System.currentTimeMillis() - now;
        log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
      } catch (Exception e) {
        log.warn("consume message failed. messageExt:{}", messageExt, e);
        context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
        return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 消费失败，让broker重发消息
      }
    }

    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 消费成功
  }
}

/**
 * @author xiejinwei02
 * @date 2022/7/14 17:21
 * 配置统一APP的RocketMQ消息发送者，RocketMQAutoConfiguration自动配置类创建的RocketMQTemplate Bean是大公共消息平台的RocketMQ消息发送者
 * 所以存在两个不同RocketMQ集群的消息发送者
 */
@Configuration
//@ConditionalOnExpression("${push.enabled:false}==true")
@EnableConfigurationProperties(RocketMQProperties.class)
public class MqConfig {
    @Value("${rocketmq-app.name-server}")
    String nameServer;

    @Value("${rocketmq-app.producer.group}")
    String groupName;

    /**
     * 统一APP RocketMQ 消息发送模板类
     * @param rocketMQMessageConverter
     * @param rocketMQProperties
     * @return
     */
    @Bean
    public RocketMQTemplate appRocketMQTemplate(@Autowired RocketMQMessageConverter rocketMQMessageConverter,RocketMQProperties rocketMQProperties){
        boolean isEnableMsgTrace = true;
        String customizedTraceTopic = MixAll.RMQ_SYS_TRACE_TOPIC;
        RocketMQTemplate appRocketMQTemplate = new RocketMQTemplate();
        DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, "", "", isEnableMsgTrace, customizedTraceTopic); // 参照自动配置类创建发送者
        producer.setNamesrvAddr(nameServer);
        RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
        producer.setInstanceName("AppMQProducer"); // 必须设置实例名称，根据实际业务名称设置
        appRocketMQTemplate.setProducer(producer);
        appRocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        return appRocketMQTemplate;
    }

    /**
     * 待办线程池，用于异步设置“待办列表更新”期望值、推送待办APP消息
     * @return
     */
    @Bean
    public ThreadPoolExecutor todoThreadPool(){
        return new ThreadPoolExecutor(20,
                200,
                300,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(10000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }
}

@Slf4j
@Service
public class MessageServiceImpl implements MessageService {
    @Autowired
    @Qualifier("rocketMQTemplate")
    RocketMQTemplate rocketMQTemplate; // 第一个业务rocketmq集群的消息发送模板类Bean
    
    @Resource
    RocketMQTemplate rocketMQTemplate; // 第一个业务rocketmq集群的消息发送模板类Bean
    
    @Autowired
    @Qualifier("appRocketMQTemplate")
    RocketMQTemplate appRocketMQTemplate; // 第二个业务rocketmq集群的消息发送模板类Bean
    
    @Resource
  	RocketMQTemplate appRocketMQTemplate;  // 第二个业务rocketmq集群的消息发送模板类Bean  
    
    private void sendRetryPushTodo(DeviceInfoRetDTO device,TodoEntity todoEntity){
        RePushTodoEntity rePushTodo = new RePushTodoEntity();
        rePushTodo.setSourceAppName(todoEntity.getSourceAppName());
        rePushTodo.setSubject(todoEntity.getSubject());
        rePushTodo.setTitle(todoEntity.getTitle());
        rePushTodo.setModelId(todoEntity.getModelId());
        rePushTodo.setMobileLink(todoEntity.getMobileLink());
        rePushTodo.setUserAccount(device.getUserAccount());
        rePushTodo.setUserType(device.getUserType().getText());
        rePushTodo.setSystemName(device.getOs().name().toLowerCase());
        rePushTodo.setDeviceId(device.getPlatformDeviceId());
        rePushTodo.setPushAppName(device.getAppName());
        rePushTodo.setExtra(todoEntity.getExtra());
        // 同步发送
        SendResult sendResult = appRocketMQTemplate.syncSend(CommonConstant.RETRY_PUSH_TODO, MessageBuilder.withPayload(rePushTodo).setHeader(RocketMQHeaders.KEYS, todoEntity.getModelId()).build());
        if(!sendResult.getSendStatus().equals(SendStatus.SEND_OK)){
            log.error("发送MQ消息，重推APP待办消息提醒，失败,{}",rePushTodo);
        }
    }
}

@Slf4j
@Service
@RocketMQMessageListener(topic = "Employee-Todo",selectorExpression = "*",consumerGroup = "EmployeeTodo-PT20090",consumeMode = ConsumeMode.ORDERLY)
public class EmployeeTodoListener extends BaseTodoListener implements RocketMQListener<MessageExt> {

    @SneakyThrows
    @Override
    public void onMessage(MessageExt payload) {
        MessageBody.TodoBody todoBody = convertTodoBody(payload);
        handleTodo(todoBody,payload.getTags(), UserTypeEnum.BIP.getText());
    }
}

/**
 * @author xiejinwei02
 * @date 2022/9/27 16:44
 * 降低消费线程数，避免高并发调用总线接口
 */
@Slf4j
@Service
@RocketMQMessageListener(topic = "${topic.todo}",selectorExpression = CommonConstant.TAG_DONE,nameServer = "${rocketmq-app.name-server:}",consumerGroup = "${consumerGroup.notifyTodoDone}",consumeThreadMax = 10,accessKey = "access",secretKey = "secret")
public class NotifyTodoDoneListener implements RocketMQListener<SetTodoDoneEntity> {
    @Autowired
    MessageService messageService;

    @SneakyThrows
    @Override
    public void onMessage(SetTodoDoneEntity entity) {
        messageService.notifyBipSetTodoDone(entity);
    }
}

public static RPCHook getRPCHookByAkSk(Environment env, String accessKeyOrExpr, String secretKeyOrExpr) {
    String ak, sk;
    try {
        ak = env.resolveRequiredPlaceholders(accessKeyOrExpr);
        sk = env.resolveRequiredPlaceholders(secretKeyOrExpr);
    } catch (Exception e) {
        // Ignore it
        ak = null;
        sk = null;
    }
    // 不为空才创建rpc连接
    if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
        return new AclClientRPCHook(new SessionCredentials(ak, sk));
    }
    return null;
}

package org.apache.rocketmq.spring.core;

public interface RocketMQListener<T> {
    void onMessage(T message);
}

package org.apache.rocketmq.spring.core;

/**
 * The consumer supporting request-reply should implement this interface.
 *
 * @param <T> the type of data received by the listener
 * @param <R> the type of data replying to producer
 */
public interface RocketMQReplyListener<T, R> {
    /**
     * @param message data received by the listener
     * @return data replying to producer
     */
    R onMessage(T message);
}

public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {

    @SuppressWarnings("unchecked")
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt messageExt : msgs) {
            log.debug("received msg: {}", messageExt);
            try {
                long now = System.currentTimeMillis();
                handleMessage(messageExt);
                long costTime = System.currentTimeMillis() - now;
                log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
            } catch (Exception e) {
                log.warn("consume message failed. messageExt:{}, error:{}", messageExt, e);
                context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

private void handleMessage(
    MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {
    if (rocketMQListener != null) {
        rocketMQListener.onMessage(doConvertMessage(messageExt));
    } else if (rocketMQReplyListener != null) {
        // 接收返回值
        Object replyContent = rocketMQReplyListener.onMessage(doConvertMessage(messageExt));
        // 创建MQ消息
        Message<?> message = MessageBuilder.withPayload(replyContent).build();
        // 第一个参数是消费者的入参消息
        // 第二个参数是消费者的出参
        org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, convertToBytes(message));
        // 异步发送消息
        consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(replyMessage, new SendCallback() {
            @Override public void onSuccess(SendResult sendResult) {
                if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                    log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
                } else {
                    log.info("Consumer replies message success.");
                }
            }

            @Override public void onException(Throwable e) {
                log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
            }
        });
    }
}

public class MessageUtil {
    public static Message createReplyMessage(final Message requestMessage, final byte[] body) throws MQClientException {
        if (requestMessage != null) {
            Message replyMessage = new Message();
            String cluster = requestMessage.getProperty(MessageConst.PROPERTY_CLUSTER);
            String replyTo = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
            String correlationId = requestMessage.getProperty(MessageConst.PROPERTY_CORRELATION_ID);
            String ttl = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_TTL);
            replyMessage.setBody(body);
            if (cluster != null) {
                // 设置返回消息的topic
                String replyTopic = MixAll.getReplyTopic(cluster);
                replyMessage.setTopic(replyTopic);
                MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_TYPE, MixAll.REPLY_MESSAGE_FLAG);
                MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_CORRELATION_ID, correlationId);
                MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, replyTo);
                MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_TTL, ttl);

                return replyMessage;
            } else {
                throw new MQClientException(ClientErrorCode.CREATE_REPLY_MESSAGE_EXCEPTION, "create reply message fail, requestMessage error, property[" + MessageConst.PROPERTY_CLUSTER + "] is null.");
            }
        }
        throw new MQClientException(ClientErrorCode.CREATE_REPLY_MESSAGE_EXCEPTION, "create reply message fail, requestMessage cannot be null.");
    }

    public static String getReplyToClient(final Message msg) {
        return msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
    }
}

String cluster = requestMessage.getProperty(MessageConst.PROPERTY_CLUSTER);  // 常量值=CLUSTER
String replyTopic = MixAll.getReplyTopic(cluster);
replyMessage.setTopic(replyTopic);

    public static String getReplyTopic(final String clusterName) {
        return clusterName + "_" + REPLY_TOPIC_POSTFIX; // 常量值=REPLY_TOPIC
    }

/**
 * @author xiejinwei02
 * @date 2023/3/8 8:49
 * 设备信息上报异步更新消费者
 */
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
@RocketMQMessageListener(topic = "${rocketmq.topic.device_info:TOPIC_DEVICE_INFO}",selectorExpression = "${rocketmq.tag.reported:TAG_REPORTED}",consumerGroup = "${rocketmq.consumerGroup.deviceInfoReported:CG_DEVICE_INFO_REPORTED}")
public class DeviceInfoReportedListener implements RocketMQListener<DeviceInfoReportVO>, RocketMQPushConsumerLifecycleListener {
	private final DeviceNativeService deviceNativeService;
	
	@Override
	public void onMessage(DeviceInfoReportVO message) {
		try {
			deviceNativeService.reportedDevice(message);
		} catch (Exception e) {
			// 记录异常后不重新消费
			log.error("设备信息上报MQ消费失败:{},异常:{}",message,e);
		}
	}
	
	@Override
	public void prepareStart(DefaultMQPushConsumer consumer) {
		consumer.setMaxReconsumeTimes(1);
	}
}

public interface RocketMQPushConsumerLifecycleListener extends RocketMQConsumerLifecycleListener<DefaultMQPushConsumer> {
}

public interface RocketMQConsumerLifecycleListener<T> {
    void prepareStart(final T consumer);
}