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

男娘i 2023-09-29 20:59 24阅读 0赞

生产者：

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>

![5544feac801774abddae0788894af70e.png][]

yml文件：

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

![3e06411d6060abcf8b3d54e14c8fc39e.png][]

![56fb4243b30db5ecfc0442cad1659286.png][]

![748915b9f4f40c675311111248309c9d.png][]

![522d1916425343621c2b4c698d843d28.png][]

![7ebc2fd4676db98f903c41064b92605f.png][]

设置config：

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);
    
        }
    
    
    }

![cfd7be25ab14e605489a9a407ef68162.png][]

注入监听器：

package org.example.config;
    
    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中
        @Autowired
        RabbitTemplate rabbitTemplate;
    
    
        @PostConstruct
        public void init()
        {
            rabbitTemplate.setConfirmCallback(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);
            }
    
        }
    }

![08e0d0c0516e34e6e432b7280ec6b59f.png][]

![833a4c73f16cec1cec2c1728360ab767.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);
        }
    
    }

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

![72f4a59f1bbd6b7b0fc949788dcb961e.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);
    
        }
    }

![eb23018543c7502e82b75133b14714d2.png][]

![bffd8fce35def175661f9453322445e9.png][]

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

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

![ee3961971eb2a02b34cec0e17e986f81.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.#");
        }
    }

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

![2481963332b68c49f7dad9a6c730e15d.png][]

启动服务：

![87ef88fec0f13402a1302eb4dfb746d6.png][]

![dacd073e15c43cbeb0c1a1a5a545ccdd.png][]

[5544feac801774abddae0788894af70e.png]: https://img-blog.csdnimg.cn/img_convert/5544feac801774abddae0788894af70e.png
[3e06411d6060abcf8b3d54e14c8fc39e.png]: https://img-blog.csdnimg.cn/img_convert/3e06411d6060abcf8b3d54e14c8fc39e.png
[56fb4243b30db5ecfc0442cad1659286.png]: https://img-blog.csdnimg.cn/img_convert/56fb4243b30db5ecfc0442cad1659286.png
[748915b9f4f40c675311111248309c9d.png]: https://img-blog.csdnimg.cn/img_convert/748915b9f4f40c675311111248309c9d.png
[522d1916425343621c2b4c698d843d28.png]: https://img-blog.csdnimg.cn/img_convert/522d1916425343621c2b4c698d843d28.png
[7ebc2fd4676db98f903c41064b92605f.png]: https://img-blog.csdnimg.cn/img_convert/7ebc2fd4676db98f903c41064b92605f.png
[cfd7be25ab14e605489a9a407ef68162.png]: https://img-blog.csdnimg.cn/img_convert/cfd7be25ab14e605489a9a407ef68162.png
[08e0d0c0516e34e6e432b7280ec6b59f.png]: https://img-blog.csdnimg.cn/img_convert/08e0d0c0516e34e6e432b7280ec6b59f.png
[833a4c73f16cec1cec2c1728360ab767.png]: https://img-blog.csdnimg.cn/img_convert/833a4c73f16cec1cec2c1728360ab767.png
[f91595bf99866f9d97489b98b36e5f23.png]: https://img-blog.csdnimg.cn/img_convert/f91595bf99866f9d97489b98b36e5f23.png
[72f4a59f1bbd6b7b0fc949788dcb961e.png]: https://img-blog.csdnimg.cn/img_convert/72f4a59f1bbd6b7b0fc949788dcb961e.png
[eb23018543c7502e82b75133b14714d2.png]: https://img-blog.csdnimg.cn/img_convert/eb23018543c7502e82b75133b14714d2.png
[bffd8fce35def175661f9453322445e9.png]: https://img-blog.csdnimg.cn/img_convert/bffd8fce35def175661f9453322445e9.png
[b662dd7993f382a39e4d40af3edf1428.png]: https://img-blog.csdnimg.cn/img_convert/b662dd7993f382a39e4d40af3edf1428.png
[aa51dea2648e439189bd5f962acb7469.png]: https://img-blog.csdnimg.cn/img_convert/aa51dea2648e439189bd5f962acb7469.png
[ee3961971eb2a02b34cec0e17e986f81.png]: https://img-blog.csdnimg.cn/img_convert/ee3961971eb2a02b34cec0e17e986f81.png
[b620c1d390a43c295e36cfe476cecaec.png]: https://img-blog.csdnimg.cn/img_convert/b620c1d390a43c295e36cfe476cecaec.png
[2481963332b68c49f7dad9a6c730e15d.png]: https://img-blog.csdnimg.cn/img_convert/2481963332b68c49f7dad9a6c730e15d.png
[87ef88fec0f13402a1302eb4dfb746d6.png]: https://img-blog.csdnimg.cn/img_convert/87ef88fec0f13402a1302eb4dfb746d6.png
[dacd073e15c43cbeb0c1a1a5a545ccdd.png]: https://img-blog.csdnimg.cn/img_convert/dacd073e15c43cbeb0c1a1a5a545ccdd.png