java——spring boot集成RabbitMQ——高级特效——可靠性投递总结——topic——路由模式——代码汇总...

喜欢ヅ旅行 2023-09-29 20:59 3阅读 0赞

可靠性在3个方面：

1、发送到到rmq——confirm机制；

2、交换机到队列——return机制；

3、队列到消费者——自动ack改为手动ack；

![be8a2514f5e8b1d341a7dbf7764729fb.png][]

![bd9557e68a4cab832409265878773ca2.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>

![b4cedf7190ed97969335bddc9e049cbb.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

![59f94b194e2e833aecca6e936bc5fe6a.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() + "没有成功投递到队列");
        }
    }

![a2465274c7b1f830ff5b0dbe804d8735.png][]

![02ce3a3af814d69fb0344e45e2aa6d49.png][]

![70b829d3aaf334466c3ae77b815d40e7.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);
    
        }
    
    
    }

![82299ef169fb7e244c88f11e605cd40b.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","huawei.a",info,correlationData);
        }
    
    }

![b80d9c4b2686d9bf30c2bb0822769200.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";
        }
    
    }

![0ff7ef937dba50df4163432a9f852322.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);
    
        }
    }

![09597286cd4aa4d05eae33769f7ef780.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>springrmqtopicreceiver</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>

![1aae4d1de4dd903da10699a6b7a075df.png][]

yml文件：

server:
      port: 8081
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        #关闭自动ack，设置为手动ack
        listener:
          simple:
            acknowledge-mode: manual

![1e531be4485d9819ffb2c5efe4ce848f.png][]

配置文件：

package org.example.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    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);
        }
    
        @Bean
        Queue xiaomi(){
            return new Queue("xiaomi",true);
        }
    
        @Bean
        Queue huawei(){
            return new Queue("huawei",true);
        }
    
        @Bean
        Binding xiaomiBinding(){
            //xiaomi.#：表示消息的routingKey是以xiaomi开头的就会路由到xiaomi的队列
            return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
        }
    
        @Bean
        Binding huaweiBinding(){
            return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
        }
    }

![04d0e05313c471d49796d303bd0b3c2c.png][]

接收消息：

package org.example.receiver;
    
    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;
    
    
    @Component
    public class TopicReceiver {
    
        //分别监听名称为xiaomi、huawei的队列
        @RabbitListener(queues = "xiaomi")
        public void handlerXM(Message message,String msg, Channel channel) throws IOException {
    
            System.out.println("小米："+msg);
    
            //手动签收，不启动批量签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    
            System.out.println(message.getMessageProperties().getDeliveryTag());
        }
    
    
        @RabbitListener(queues = "huawei")
        public void handlerHW(Message message,String msg, Channel channel) throws IOException {
    
    
            System.out.println("华为："+msg);
    
            //手动签收，不启动批量签收
            //告诉rmq签收的消息的id。以及是否批量签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    
        }
    }

![7ff07ce850b2a87b708de6ebcec6c8cb.png][]

启动服务：

![b107fb0c8b37abf249f3cdf862681461.png][]

[be8a2514f5e8b1d341a7dbf7764729fb.png]: https://img-blog.csdnimg.cn/img_convert/be8a2514f5e8b1d341a7dbf7764729fb.png
[bd9557e68a4cab832409265878773ca2.png]: https://img-blog.csdnimg.cn/img_convert/bd9557e68a4cab832409265878773ca2.png
[b4cedf7190ed97969335bddc9e049cbb.png]: https://img-blog.csdnimg.cn/img_convert/b4cedf7190ed97969335bddc9e049cbb.png
[59f94b194e2e833aecca6e936bc5fe6a.png]: https://img-blog.csdnimg.cn/img_convert/59f94b194e2e833aecca6e936bc5fe6a.png
[a2465274c7b1f830ff5b0dbe804d8735.png]: https://img-blog.csdnimg.cn/img_convert/a2465274c7b1f830ff5b0dbe804d8735.png
[02ce3a3af814d69fb0344e45e2aa6d49.png]: https://img-blog.csdnimg.cn/img_convert/02ce3a3af814d69fb0344e45e2aa6d49.png
[70b829d3aaf334466c3ae77b815d40e7.png]: https://img-blog.csdnimg.cn/img_convert/70b829d3aaf334466c3ae77b815d40e7.png
[82299ef169fb7e244c88f11e605cd40b.png]: https://img-blog.csdnimg.cn/img_convert/82299ef169fb7e244c88f11e605cd40b.png
[b80d9c4b2686d9bf30c2bb0822769200.png]: https://img-blog.csdnimg.cn/img_convert/b80d9c4b2686d9bf30c2bb0822769200.png
[0ff7ef937dba50df4163432a9f852322.png]: https://img-blog.csdnimg.cn/img_convert/0ff7ef937dba50df4163432a9f852322.png
[09597286cd4aa4d05eae33769f7ef780.png]: https://img-blog.csdnimg.cn/img_convert/09597286cd4aa4d05eae33769f7ef780.png
[1aae4d1de4dd903da10699a6b7a075df.png]: https://img-blog.csdnimg.cn/img_convert/1aae4d1de4dd903da10699a6b7a075df.png
[1e531be4485d9819ffb2c5efe4ce848f.png]: https://img-blog.csdnimg.cn/img_convert/1e531be4485d9819ffb2c5efe4ce848f.png
[04d0e05313c471d49796d303bd0b3c2c.png]: https://img-blog.csdnimg.cn/img_convert/04d0e05313c471d49796d303bd0b3c2c.png
[7ff07ce850b2a87b708de6ebcec6c8cb.png]: https://img-blog.csdnimg.cn/img_convert/7ff07ce850b2a87b708de6ebcec6c8cb.png
[b107fb0c8b37abf249f3cdf862681461.png]: https://img-blog.csdnimg.cn/img_convert/b107fb0c8b37abf249f3cdf862681461.png