RabbitMQ ACK入门

朴灿烈づ我的快乐病毒、 2022-09-13 01:59 103阅读 0赞

# jar包 #

lombok,amqp,logback

<dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.3</version>
    </dependency>

## 配置文件 ##

server:
      port: 9090
    spring:
      rabbitmq:
        publisher-confirms: true
        host: 192.168.40.130
        port: 5672
        username: admin
        password: admin
        virtual-host: /demo

## 手动ack需要的实现类 ##

package com.zfh.config.callback;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class ExchangeCallBack implements RabbitTemplate.ConfirmCallback {
        protected final static Logger logger =LoggerFactory.getLogger(ExchangeCallBack.class);
        /**
         * 发布者确认的回调。
         *
         * @param correlationData 回调的相关数据。
         * @param ack             ack为真，nack为假
         * @param cause           一个可选的原因，用于nack，如果可用，否则为空。
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
          if (correlationData==null){
              logger.error("发送消息到MQ时未发送关联data到confirmCallback，请检查发送逻辑！");
              throw new IllegalArgumentException("发送消息到MQ时未发送关联data到confirmCallback，请检查发送逻辑！");
          }
          if (ack){
              logger.info("Exchange收到消息");
          }
        }
    }
    
    
    
    
    
    
    
    
    
    
    package com.zfh.config.callback;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import java.util.UUID;
    
    @Component
    @Slf4j
    public class QueueReturnBack implements RabbitTemplate.ReturnCallback{
        /*
       使用Jackson2JsonMessageConverter处理器，客户端发送JSON类型数据，但是没有指定消息的contentType类型，
       那么Jackson2JsonMessageConverter就会将消息转换成byte[]类型的消息进行消费。
       如果指定了contentType为application/json，那么消费端就会将消息转换成Map类型的消息进行消费。
       如果指定了contentType为application/json，并且生产端是List类型的JSON格式，那么消费端就会将消息转换成
       List类型的消息进行消费。
         */
        private Jackson2JsonMessageConverter jackson2JsonMessageConverter;
    
        @PostConstruct
        public void initConverter(){
            this.jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        }
    
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            Object o = jackson2JsonMessageConverter.fromMessage(message);
            String s = UUID.randomUUID().toString();
            log.error("交换机‘【{}】’路由到队列中失败！请检查MQ配置！路由key为：【{}】", exchange, routingKey);
    
        }
    }

## config配置类 ##

package com.zfh.config;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.zfh.config.callback.ExchangeCallBack;
    import com.zfh.config.callback.QueueReturnBack;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.Connection;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.PostConstruct;
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.TimeoutException;
    
    @Configuration
    public class RabbitMQConfig {
    
    
           @Autowired
           private RabbitTemplate rabbitTemplate;
    
            @Autowired
           private ExchangeCallBack exchangeCallBack;
            @Autowired
           private QueueReturnBack queueReturnBack;
    
        //创建一个与MQ的连接
        @Autowired
        CachingConnectionFactory connectionFactory;
    
    
        @PostConstruct
        public void init() throws IOException, TimeoutException {
            //获取连接，获取信道 ，创建与交换机的通道，每个通道代表一个会话
            Connection connection = this.connectionFactory.createConnection();
            Channel channel = connection.createChannel(false);
            //声明交换机 String exchange, BuiltinExchangeType type
            /**
             * 参数明细
             * 1、交换机名称
             * 2、交换机类型，fanout、topic、direct、headers
             */
            channel.exchangeDeclare("DEMO1", BuiltinExchangeType.TOPIC,true,false,null);
            channel.exchangeDeclare("DEMO2", BuiltinExchangeType.TOPIC,true,false,null);
    
            //队列定义，声明队列
            Map<String, Object> args = new HashMap<>();
            args.put("dead-demo","DEMO1_DEAD");
            args.put("dead-demow","DEMO2_DEAD");
    
            // (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String,Object> arguments)
            /**
             * 参数明细：
             * 1、队列名称
             * 2、是否持久化
             * 3、是否独占此队列
             * 4、队列不用是否自动删除
             * 5、参数
             */
            channel.queueDeclare("QUEUEDEMO1",true,false,false,args);
            channel.queueDeclare("QUEUEDEMO2",true,false,false,args);
    
            //交换机和队列绑定String queue, String exchange, String routingKey
            /**
             * 参数明细
             * 1、队列名称
             * 2、交换机名称
             * 3、路由key
             */
            channel.queueBind("QUEUEDEMO1","DEMO1","demo1.test");
            channel.queueBind("QUEUEDEMO2","DEMO2","demo2.test");
    
    
            //关闭信道
            channel.close();
    
    
        }
        @PostConstruct
        public void initRabbitTemplateCallback() {
            this.rabbitTemplate.setConfirmCallback(exchangeCallBack);
            this.rabbitTemplate.setReturnCallback(queueReturnBack);
        }
    
        @Bean
        public Jackson2JsonMessageConverter messageConverter() {
            return new Jackson2JsonMessageConverter();
        }
    }

## 测试生产类 ##

package com.zfh;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @SpringBootTest(classes=StartApplication.class)
    @RunWith(SpringRunner.class)
    public class RabbitMqTest {
        @Autowired
        private RabbitTemplate rabbitTemplate;
        @Test
        public void testhelloworld() {
            //第一个参数：发送的队列 第二个参数： 发送的信息
            rabbitTemplate.convertAndSend("hellosr","你妈的的学习 rabbitmq");
        }
        @Test
        public  void AckTest(){
    
            CorrelationData data = new CorrelationData();
            data.setId("1");
            rabbitTemplate.convertAndSend("DEMO1","demo1.test","叼哦",data);
        }
    
    }

## 消费者测试 ##

@Component
    @RabbitListener(queuesToDeclare = @Queue("hellosr"))
    public class UserTest {
        @RabbitHandler
        public void receivel(String message) {
            System.out.println(message);
        }
    }