java——spring boot集成RabbitMQ——高级特效——可靠性投递——spring boot实现return机制...

川长思鸟来 2023-09-29 21:00 5阅读 0赞

交换机到队列的可靠性保证机制：

![f7f979eb5a9c4664e7ec3835f3e4d5df.png][]

![08b1d86d9dc142daf82ae6f3441e9143.png][]

![813b37ee64cfc73a7235155e455647c9.png][]

![32f5ca22ff3afde426d3da9bd99f96fa.png][]

![a85a1cef04eaff3eb6d87b03cf4637ff.png][]

![aab5243a681e3f07f779f8838a33c19a.png][]

![c30d6e25dd133e4f119ff2694f63df9d.png][]

====================================================================================

pom文件：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>springrmqtopicsender</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <maven.compiler.source>8</maven.compiler.source>
            <maven.compiler.target>8</maven.compiler.target>
        </properties>
    
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.4.5</version>
            <relativePath/>
        </parent>
    
    
        <dependencies>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-devtools</artifactId>
                <scope>runtime</scope>
                <optional>true</optional>
            </dependency>
    
        </dependencies>
    
    </project>

![d61152242f26ff97c054850d22333fde.png][]

yml文件：

server:
      port: 8082
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        publisher-confirm-type: correlated
        publisher-returns: true

![3ab2eddc9a99da6de43587c60ee9b5cd.png][]

回调配置文件：

package org.example.config;
    
    import org.springframework.amqp.core.ReturnedMessage;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    
    @Component
    public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
    
        //把监听器注入到RabbitTemplate中
        @Autowired
        RabbitTemplate rabbitTemplate;
    
    
        @PostConstruct
        public void init()
        {
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnsCallback(this);
    
        }
    
    
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    
    
            /**
             *
             * 生产者发送消息给交换机；交换机返回ack的同时，还会带一些消息云数据，云数据从correlationData中获得
             *
             * correlationData中包括：
             *
             *     private volatile String id;
             *
             *     private volatile ReturnedMessage returnedMessage;
             *
             *     ============================================================================
             *
             *     ack是个布尔值。如果生产者发送一个消息到交换机，交换机签收成功，返回true。否则返回false；
             *
             *     ================================================================================
             *
             *     cause：原因。ack返回false时，返回的具体原因
             *
             */
    
            String id = correlationData.getId();
    
            if(ack)
            {
                //消息投递成功
                System.out.println("消息投递成功：" + id);
            }
            else {
                //失败。存入到缓存中，通过定时任务，定时发送
                System.out.println("消息投递失败，原因" + cause);
            }
    
        }
    
        @Override
        public void returnedMessage(ReturnedMessage returned) {
            /**
             * 当消息没有传递到队列的时候的回调方法：
             *
             *    private final Message message;
             *
             *     private final int replyCode;
             *
             *     private final String replyText;
             *
             *     private final String exchange;
             *
             *     private final String routingKey;
             *
             */
    
            System.out.println("消息" + returned.getMessage() + "没有成功投递到队列");
        }
    }

![625c1004ece1a7fa72f111b0ec1772e6.png][]

![ae29f71c9a4bd1f78af3668b864517ae.png][]

![d4dfaa7ebf23bf020943607f382e9d24.png][]

配置交换机：

package org.example.config;
    
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * 主题交换机
     * topic策略可以根据routingKey的规则(通配符方式)进行去匹配队列进行转发规则为*.#.*
     */
    @Configuration
    public class RabbitTopicConfig
    {
    
    
        public final static String TOPIC_NAME = "amqp-topic";
    
    
        @Bean
        TopicExchange topicExchange()
        {
    
            return new TopicExchange(TOPIC_NAME,true,false);
    
        }
    
    
    }

![2518027ba635266872696cafcbfb3e1e.png][]

发送消息：

package org.example.sender;
    
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.UUID;
    
    /**
     * 消息生产者 发送消息
     */
    @Component
    public class MessageSender {
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        /**
         * 发送消息
         * @param info
         */
        public void send(String info)
        {
    
            System.out.println("发送消息>>>"+info);
    
            CorrelationData correlationData = new CorrelationData();
    
            String uuid = UUID.randomUUID().toString();
            System.out.println(uuid);
    
            correlationData.setId(uuid);
    
            rabbitTemplate.convertAndSend("amqp-topic","ahuawei.a",info,correlationData);
        }
    
    }

![945ec187726d3619d868f7264fc0e7c7.png][]

发送：

package org.example.controller;
    
    import org.example.sender.MessageSender;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @Auther: moerhai@qq.com
     * @Date: 2020/10/4 11:34
     */
    @RestController
    public class IndexController {
    
        @Autowired
        MessageSender messageSender;
    
        @RequestMapping("/index")
        public String index()
        {
    
            messageSender.send("中国——路由——华为");
            return "SUCCESS";
        }
    
    }

![ccdbc4751f4dab42eaf43a907cba085c.png][]

启动服务：

package org.example;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class HelloWorldApplication
    {
    
        public static void main(String[] args)
        {
    
            SpringApplication.run(HelloWorldApplication.class, args);
    
        }
    }

![3f017637aabd92b601243dd2b55b34f9.png][]

![52286dd1302d10d2dd28f1b052a7cb84.png][]

![10ab7e3c0bdce4b4e5772721e6cc641e.png][]

[f7f979eb5a9c4664e7ec3835f3e4d5df.png]: https://img-blog.csdnimg.cn/img_convert/f7f979eb5a9c4664e7ec3835f3e4d5df.png
[08b1d86d9dc142daf82ae6f3441e9143.png]: https://img-blog.csdnimg.cn/img_convert/08b1d86d9dc142daf82ae6f3441e9143.png
[813b37ee64cfc73a7235155e455647c9.png]: https://img-blog.csdnimg.cn/img_convert/813b37ee64cfc73a7235155e455647c9.png
[32f5ca22ff3afde426d3da9bd99f96fa.png]: https://img-blog.csdnimg.cn/img_convert/32f5ca22ff3afde426d3da9bd99f96fa.png
[a85a1cef04eaff3eb6d87b03cf4637ff.png]: https://img-blog.csdnimg.cn/img_convert/a85a1cef04eaff3eb6d87b03cf4637ff.png
[aab5243a681e3f07f779f8838a33c19a.png]: https://img-blog.csdnimg.cn/img_convert/aab5243a681e3f07f779f8838a33c19a.png
[c30d6e25dd133e4f119ff2694f63df9d.png]: https://img-blog.csdnimg.cn/img_convert/c30d6e25dd133e4f119ff2694f63df9d.png
[d61152242f26ff97c054850d22333fde.png]: https://img-blog.csdnimg.cn/img_convert/d61152242f26ff97c054850d22333fde.png
[3ab2eddc9a99da6de43587c60ee9b5cd.png]: https://img-blog.csdnimg.cn/img_convert/3ab2eddc9a99da6de43587c60ee9b5cd.png
[625c1004ece1a7fa72f111b0ec1772e6.png]: https://img-blog.csdnimg.cn/img_convert/625c1004ece1a7fa72f111b0ec1772e6.png
[ae29f71c9a4bd1f78af3668b864517ae.png]: https://img-blog.csdnimg.cn/img_convert/ae29f71c9a4bd1f78af3668b864517ae.png
[d4dfaa7ebf23bf020943607f382e9d24.png]: https://img-blog.csdnimg.cn/img_convert/d4dfaa7ebf23bf020943607f382e9d24.png
[2518027ba635266872696cafcbfb3e1e.png]: https://img-blog.csdnimg.cn/img_convert/2518027ba635266872696cafcbfb3e1e.png
[945ec187726d3619d868f7264fc0e7c7.png]: https://img-blog.csdnimg.cn/img_convert/945ec187726d3619d868f7264fc0e7c7.png
[ccdbc4751f4dab42eaf43a907cba085c.png]: https://img-blog.csdnimg.cn/img_convert/ccdbc4751f4dab42eaf43a907cba085c.png
[3f017637aabd92b601243dd2b55b34f9.png]: https://img-blog.csdnimg.cn/img_convert/3f017637aabd92b601243dd2b55b34f9.png
[52286dd1302d10d2dd28f1b052a7cb84.png]: https://img-blog.csdnimg.cn/img_convert/52286dd1302d10d2dd28f1b052a7cb84.png
[10ab7e3c0bdce4b4e5772721e6cc641e.png]: https://img-blog.csdnimg.cn/img_convert/10ab7e3c0bdce4b4e5772721e6cc641e.png