StringBoot集成Rabbit，根据业务返回ACK

为了维护消息的有效性，当消费消息时候处理失败时候，不进行消费，需要我们根据业务区返回ACK，本项目我使用Redis和ack机制双重保险,保障消息一定能够正确的消费

• 首先,接着上部分内容，使用Topic,机制(不明白的,可以回顾上部分内容)

• 上部分内容，我们使用SpringBoot注解,去实现，但是控制权不完全账务，当进行大规模项目时候，不太建议使用

  

   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 @RabbitListener(queues = TopicRabbitConfig.USER_QUEUE)
     
    
   

    

     

    
    

     
    @RabbitHandler
     
    
   

    

     

    
    

     

          public void processUser(String message) {
     
    
   

    

     

    
    

     
        threadPool.execute(new Runnable() {
     
    
   

    

     

    
    

     
            @Override
     
    
   

    

     

    
    

     

                  public void run() {
     
    
   

    

     

    
    

     
                logger.info("用户侧流水:{}",message);
     
    
   

    

     

    
    

     

                  }
     
    
   

    

     

    
    

     

              });
     
    
   

    

     

    
    

     

          }
     
    
  
 

• 根据源码分析，当然这里不分析源码，有兴趣的可以多失败几次就ok明白了

• 在配置类中定义监听器，监听这个序列（AcknowledgeMode.MANUAL是必须的哦）

  

   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    /**
     
    
   

    

     

    
    

     

       * 接受消息的监听，这个监听客户交易流水的消息
     
    
   

    

     

    
    

     

       * 针对消费者配置
     
    
   

    

     

    
    

     

       * @return
     
    
   

    

     

    
    

     

       */
     
    
   

    

     

    
    

     
    @Bean
     
    
   

    

     

    
    

     
    public SimpleMessageListenerContainer messageContainer1(ConnectionFactory connectionFactory, TransactionConsumeImpl transactionConsume) {
     
    
   

    

     

    
    

     
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
     
    
   

    

     

    
    

     

              container.setQueues(queueMessage());
     
    
   

    

     

    
    

     

              container.setExposeListenerChannel(true);
     
    
   

    

     

    
    

     

              container.setMaxConcurrentConsumers(1);
     
    
   

    

     

    
    

     

              container.setConcurrentConsumers(1);
     
    
   

    

     

    
    

     

              container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
     
    
   

    

     

    
    

     

              container.setMessageListener(transactionConsume);
     
    
   

    

     

    
    

     
        return container;
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     
 
     
    
  
 

  
 
 

这个 TransactionConsumeImpl 要继承ChannelAwareMessageListener，主要说的手动返回ACK就是channel。调用

  

   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      @Component
     
    
   

    

     

    
    

     

      public class TransactionConsumeImpl implements ChannelAwareMessageListener {
     
    
   

    

     

    
    

     
    private static final Logger logger = LoggerFactory.getLogger(TransactionConsumeImpl.class);
     
    
   

    

     

    
    

     
    private static final Gson gson = new Gson();
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    @Autowired
     
    
   

    

     

    
    

     

          JedisShardInfo jedisShardInfo;
     
    
   

    

     

    
    

     
    @Autowired
     
    
   

    

     

    
    

     

          ExecutorService threadPool;
     
    
   

    

     

    
    

     
    @Autowired
     
    
   

    

     

    
    

     

          BoluomeFlowService boluomeFlowService;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    @Override
     
    
   

    

     

    
    

     
    public void onMessage(Message message, Channel channel) throws Exception {
     
    
   

    

     

    
    

     
        String boby = new String(message.getBody(), "utf-8");//转换消息，我们是使用json数据格式
     
    
   

    

     

    
    

     

              threadPool.execute(new Runnable() {   //多线程处理
     
    
   

    

     

    
    

     
            @Override
     
    
   

    

     

    
    

     
            public void run() {
     
    
   

    

     

    
    

     
                Jedis jedis = jedisShardInfo.createResource();
     
    
   

    

     

    
    

     

                      jedis.sadd(TopicRabbitConfig.TRANSACTION_QUEUE, boby);//添加到key为当前消息类型的集合里面，防止丢失消息
     
    
   

    

     

    
    

     
                BoluomeFlow flow = gson.fromJson(boby, BoluomeFlow.class);
     
    
   

    

     

    
    

     
                String json = gson.toJson(flow);
     
    
   

    

     

    
    

     
                if (boluomeFlowService.insert(flow)) {  //当添加成功时候返回成功
     
    
   

    

     

    
    

     

                          logger.info("客户交易流水添加1条记录:{}", json);
     
    
   

    

     

    
    

     

                          jedis.srem(TopicRabbitConfig.TRANSACTION_QUEUE, boby);//从当前消息类型集合中移除已经消费过的消息
     
    
   

    

     

    
    

     
                    try {
     
    
   

    

     

    
    

     

                              channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手工返回ACK，通知此消息已经争取消费
     
    
   

    

     

    
    

     

                          } catch (IOException ie) {
     
    
   

    

     

    
    

     

                              logger.error("消费成功回调成功，io操作异常");
     
    
   

    

     

    
    

     

                          }
     
    
   

    

     

    
    

     

                      } else {
     
    
   

    

     

    
    

     

                          logger.info("客户交易流水添加失败记录:{}", json);
     
    
   

    

     

    
    

     

                      }
     
    
   

    

     

    
    

     

                  }
     
    
   

    

     

    
    

     

              });
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     

      }
     
    
   

    

     

    
    

     
 
     
    
  
 

  
 
 

文章来源: springlearn.blog.csdn.net，作者：西魏陶渊明，版权归原作者所有，如需转载，请联系作者。

原文链接：springlearn.blog.csdn.net/article/details/102425287