SpringBoot 整合RabbitMQ ack确认消息

女爷i 2022-09-02 14:43 287阅读 0赞

## SpringBoot 整合RabbitMQ ack确认消息 ##

废话不多说，上代码了，理论可以看之前的文章

## 版本 ##

springboot 2.x  
jdk 11  
RabbitMQ 3.8.12  
Erlang 23.2.6

## 依赖 ##

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
    
    
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

## 配置 ##

spring.application.name=springboot_rabbitmq
    spring.rabbitmq.host=localhost
    spring.rabbitmq.virtual-host=/
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.port=5672

## 定义交换器和队列以及绑定关系 ##

@Configuration
    public class RabbitmqConfig { 
    
    
        @Bean
        public Queue myQueue() { 
            return new Queue("sb.queue");
        }
    
    
        @Bean
        public Exchange myExchange() { 
    
            // new Exchange()
            // return new TopicExchange("topic.biz.ex", false, false, null);
    // return new DirectExchange("direct.biz.ex", false, false, null);
            // return new FanoutExchange("fanout.biz.ex", false, false, null);
            // return new HeadersExchange("header.biz.ex", false, false, null);
            // 交换器名称，交换器类型（），是否是持久化的，是否自动删除，交换器属性 Map集合
            // return new CustomExchange("custom.biz.ex", ExchangeTypes.DIRECT, false, false, null);
            return new DirectExchange("sb.ex", false, false, null);
        }
    
    
        @Bean
        public Binding myBining() { 
    
            // 绑定的目的地，绑定的类型：到交换器还是到队列，交换器名称，路由key， 绑定的属性
            // new Binding("", Binding.DestinationType.EXCHANGE, "", "", null);
            // 绑定的目的地，绑定的类型：到交换器还是到队列，交换器名称，路由key， 绑定的属性
            // new Binding("", Binding.DestinationType.QUEUE, "", "", null);
            // 绑定了交换器direct.biz.ex到队列myqueue，路由key是 direct.biz.ex
            return new Binding("sb.queue",
                    Binding.DestinationType.QUEUE, "sb.ex",
                    "sb.biz.ex", null);
    
        }
    
    }

## 定义生产者 ##

package com.example.demo.web;
    
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.io.UnsupportedEncodingException;
    
    @RestController
    @RequestMapping("test")
    public class ProduceController { 
    
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        @RequestMapping("/send/{message}")
        public String sendMessage(@PathVariable String message) throws UnsupportedEncodingException { 
    
            //消息属性
            MessageProperties messageProperties = MessagePropertiesBuilder.newInstance()
                    .setContentEncoding(MessageProperties.CONTENT_TYPE_JSON) //类型
                    .setHeader("hello", "hello world..").build();
            for (int i = 0; i < 10; i++) { 
                //消息编码
                Message build = MessageBuilder.withBody((message + " " + i).getBytes("utf-8")).andProperties(messageProperties).build();
                rabbitTemplate.convertAndSend("sb.ex", "sb.biz.ex", build);
            }
    
            return "ok";
        }
    
    }

连续发送10条消息

## 定义消费者 ##

消费端可以分为推消息和拉消息

推消息

@Component
    public class MyMessageListener { 
    
    // public void getMyMessage(@Payload String message, @Header(name = "hello") String value, Channel channel) { 
    // System.out.println(message);
    // System.out.println("hello = " + value);
    // }
    
    
        private Integer index = 0;
    
        // @RabbitListener(queues = "queue.boot",ackMode = "AUTO")//自动确认
        @RabbitListener(queues = "sb.queue", ackMode = "MANUAL")//手动确认
    // @RabbitListener(queues = "queue.boot",ackMode = "NONE")//不确认
        public void getMyMessage(Message message, Channel channel) throws IOException { 
            String value = message.getMessageProperties().getHeader("hello");
    
            System.out.println(message);
            //System.out.println("接收到的推送消息是: " + new String(message.getBody()));
            System.out.println("hello = " + value);
    
            final long deliveryTag = message.getMessageProperties().getDeliveryTag();
    
            if (index % 2 == 0) { 
                // 确认消息 deliveryTag表是消息唯一标识，第二个参数表示是否批量确认
                channel.basicAck(deliveryTag, false);
                System.out.println("确认消息 : " + new String(message.getBody()));
            } else { 
                // 拒收消息 第二个参数表示是否重新入列
                channel.basicReject(deliveryTag, true);
                System.out.println("拒绝消息 : " + new String(message.getBody()));
            }
            index++;
        }
    
    }

拉消息

package com.example.demo.web;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.GetResponse;
    import org.springframework.amqp.rabbit.core.ChannelCallback;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /** * 主动拉取消息，确认消息 */
    
    @RestController
    @RequestMapping("test")
    public class GetMessageController { 
    
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        @RequestMapping("/get")
        public String getMessage() { 
    
            String msg = rabbitTemplate.execute(new ChannelCallback<String>() { 
                @Override
                public String doInRabbit(Channel channel) throws Exception { 
    
                    //拉取消息
                    GetResponse getResponse = channel.basicGet("sb.queue", false);
    
                    if (getResponse==null){ 
                        return "暂无消息可消费!";
                    }
                    byte[] body = getResponse.getBody();
                    System.out.println("拉取到的消息: " + new String(body));
    
                    //确认消息 不批量确认
                    channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);
    
                    //拒绝消息 第二个参数表示不确认多个还是一个消息，最后一个参数表示不确认的消息是否重新放回队列
    // channel.basicNack(getResponse.getEnvelope().getDeliveryTag(),false,true);
                    //拒绝消息 ，并重新入列
    // channel.basicReject(getResponse.getEnvelope().getDeliveryTag(),true);
    
                    return "已经确认的消息 " + new String(body);
                }
            });
    
            return msg;
        }
    
    }

## 测试 ##

请求接口生产消息  
http://localhost:8080/test/send/test  
推送消息后台监听消费可以看到拒绝消费的消息，重新入列后，又被重新消费了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]

拉取到并且确认消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]  
也成功返回到前台，我们这儿是每次拉取并且确认一个![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/a9fa9cc75de14f2fa2dd25b010956316.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220829/af33cbd5dacd44009c97fb4d750ea186.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]: /images/20220829/2b7611cd6f3c413b9e6171a72cdc9fee.png