springboot整合rabbitmq手动签收

拼搏现实的明天。 2023-07-05 10:13 9阅读 0赞

Springboot为AMQP提供了自动化配置依赖 spring-boot-starter-amqp,因此在项目中添加该依赖

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

在application.properties中配置RabbitMQ连接信息

spring.rabbitmq.addresses=192.168.3.12
    spring.rabbitmq.port=15672
    spring.rabbitmq.username=wangsheng
    spring.rabbitmq.password=123
    #确认是否正确到达exchange，只要正确的到达exchange中，即可确认该消息返回给客户端ack
    spring.rabbitmq.publisher-confirms=true
    #确认消息是否正确到达queue,如果没有则触发，如果有则不触发
    spring.rabbitmq.publisher-returns=true
    #采用手动应答
    spring.rabbitmq.listener.simple.acknowledge-mode=manual
    #是否支持重试
    spring.rabbitmq.listener.simple.retry.enabled=true
    #最大重试次数
    spring.rabbitmq.template.retry.max-attempts=5
    #最大重试时间
    spring.rabbitmq.template.retry.max-interval=1200
    #exchange到queue失败,则回调return
    spring.rabbitmq.template.mandatory=true

配置一个队列direct

import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    @Configuration
    public class RabbitDirectConfig {
        public final static String DIRECTNAME="wangsheng-direct";
        @Bean
        public Queue queue() {
            /** 参数含义
             * 第一个是队列名
             *  第二个 durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
             *  第三个exclusive  表示该消息队列是否只在当前connection生效,默认是false
             *  第四个auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
             */
            return new Queue("hello",true,false,false);// 使用DirectExchange时，routingkey与队列名相同，只配置一个Queue即可，DirectExchange和Binging两个的Bean配置可以省略。
    
        }
    }

配置一个生产者

import com.alibaba.fastjson.JSONObject;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageBuilder;
    import org.springframework.amqp.core.MessageDeliveryMode;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    import java.util.UUID;
    /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class publisher {
        @Autowired
    RabbitTemplate rabbitTemplate;
        //回调函数: confirm确认
        final  RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            /**
             * CorrelationData 消息的附加信息，即自定义id
             * isack 代表消息是否被broker（MQ）接收 true 代表接收 false代表拒收。
             * cause 如果拒收cause则说明拒收的原因，帮助我们进行后续处理
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("correlationData-----------" + correlationData);
                System.out.println("ack-----------" + ack);
                if(ack){
                    //更新数据库，可靠性投递机制
                    System.err.println("消息已确认....");
                }else {
                    System.err.println(cause);
    
    
                }
            }
        };
        //回调函数: return返回
        final  RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.err.println("return exchange: " + exchange + ", routingKey: "
                        + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
    
            }
        };
    
        //发送消息方法调用: 构建Message消息
        @Test
        public void send() throws Exception {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("email", "wangsheng@163.com");
            jsonObject.put("timestamp", System.currentTimeMillis());
            String jsonString = jsonObject.toJSONString();
            String orderId = UUID.randomUUID().toString();
            Message messages = MessageBuilder.withBody(jsonString.getBytes())
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                    .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
                    .setMessageId(orderId)
                    .build();
            CorrelationData correlationData = new CorrelationData(orderId);
            rabbitTemplate.setConfirmCallback(confirmCallback);
            rabbitTemplate.setReturnCallback(returnCallback);
           rabbitTemplate.convertAndSend("hello",messages,correlationData);//routingkey与队列名相同都是hello
    
        }
    }

配置一个消费者

import com.alibaba.fastjson.JSONObject;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import java.io.IOException;
    /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    @Component
    public class DirectReceiver {
        @RabbitListener(queues="hello")//监听队列hello
        public void handler1(Message message, Channel channel) throws IOException {
            String msg = new String(message.getBody(), "UTF-8");
            String messageId = message.getMessageProperties().getMessageId();
            JSONObject jsonObject = JSONObject.parseObject(msg);
            try {
                    //消息确认
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                System.out.println("消费消息：" + jsonObject);
            }catch (Exception e){
                //失败后消息被确认,
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                // channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);失败后消息重新放回队列
                // 丢弃该消息
                //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                System.out.println("签收失败:" + jsonObject);
            }
        }
    }

发送消息测试，签收成功

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xldHRlcnNz_size_16_color_FFFFFF_t_70][]

RabbitMQ中一共提供了4种不同的Exchange策略，分别是Direct、Fanout、Topic以及Header,这里只举例Direct，其他的几种策略参考[https://blog.csdn.net/letterss/article/details/102932893][https_blog.csdn.net_letterss_article_details_102932893]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xldHRlcnNz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200219160942625.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xldHRlcnNz,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_letterss_article_details_102932893]: https://blog.csdn.net/letterss/article/details/102932893