RabbitMQ -springboot整合rabbitmq

Love The Way You Lie 2021-06-10 20:39 683阅读 0赞

课程：[https://www.cnblogs.com/guchunchao/p/13173406.html][https_www.cnblogs.com_guchunchao_p_13173406.html]

**阅读目录**

*  [配置生产者][Link 1]
 *  [配置消费者][Link 2]

--------------------

导入pom.xml

![961ddebeb323a10fe0623af514929fc1.png][]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.3.1.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.everjiankang</groupId>
        <artifactId>rabbitmqDemo</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>rabbitmqDemo</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-devtools</artifactId>
                <scope>runtime</scope>
                <optional>true</optional>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-configuration-processor</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.71</version>
            </dependency>
            
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
            </dependency>
            
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    </project>

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

[回到顶部][Link 3]

## 配置生产者 ##

第一步：配置application.properties

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

spring.rabbitmq.addresses=127.0.0.1:5672
    spring.rabbitmq.username=xiaochao
    spring.rabbitmq.password=root
    spring.rabbitmq.virtual-host=/
    spring.rabbitmq.connection-timeout=15000
    
    #生产端配置
    #开启发送确认,此配置在Springboot2.3.0版本中已经@Deprecated了，默认就是
    # spring.rabbitmq.publisher-confirms=true
    #
    spring.rabbitmq.publisher-confirm-type=simple
    #开启发送失败退回
    spring.rabbitmq.publisher-returns=true
    #开启执行return回调
    spring.rabbitmq.template.mandatory=true

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第二步：开发启动类Application.java

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

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

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第三步：加入实体类order

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

package com.everjiankang.dependency.model;
    
    import java.io.Serializable;
    
    public class Order implements Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 1L;
        
        private String id;
        private String name;
        
        public Order() {
        }
    
        public Order(String id, String name) {
            super();
            this.id = id;
            this.name = name;
        }
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        } 
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第四步：开发消息发送方法

（需要先手动创建交换器、队列、绑定关系。或者通过代码Build）

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

package com.everjiankang.dependency.springboot.producer;
    
    import java.util.Map;
    
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
    import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Component;
    
    import com.everjiankang.dependency.model.Order;
    
    @Component
    public class RabbitmqSender {
        @Autowired
        private RabbitTemplate rabbitTemplate;
        
        /**
         * publisher-confirms,实现一个监听器用于监听broker端给我们返回的确认请求：
         * RabbitTemplate.ConfirmCallback
         * 
         * publisher-returns，保证消息对broker端是可达的，如果出现路由键不可达的情况，
         *     则使用监听器对不可达的消息进行后续的处理，保证消息的路由成功：
         * RabbitTemplate.ReturnCallback
         * 
         * 注意：在发送消息的时候对template配置mandatory=true保证ReturnCallback监听有效
         * 生产端还可以配置其他属性，比如发送重试，超时时间、次数、间隔等。
         */
        final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.err.println("confirm correlationData:" + correlationData);
                System.err.println("confirm ack:" + ack);
                if(!ack) {
                    System.err.println("异常处理。。。");
                }
            }
        };
        
        final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
            
            @Override
            public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
                    String exchange, String routingKey) {
                System.err.println("return exchange:" + exchange);
                System.err.println("return routingKey:" + routingKey);
                System.err.println("return replyCode:" + replyCode);
                System.err.println("return replyText:" + replyText);
            }
        };
        
        public void send(Object message, Map<String, Object> properties) throws Exception {
            MessageHeaders mhs = new MessageHeaders(properties);
            Message msg = MessageBuilder.createMessage(message, mhs);
            rabbitTemplate.setConfirmCallback(confirmCallback);
            rabbitTemplate.setReturnCallback(returnCallback);
            CorrelationData correlationData = new CorrelationData();
            //实际消息的唯一id
            correlationData.setId("dwz123456");//id + 时间戳 （必须是全局唯一的）
            rabbitTemplate.convertAndSend("exchange-1", "spring.abc", msg, correlationData);
        }
        
        public void sendOrder(Order order) throws Exception {
            rabbitTemplate.setConfirmCallback(confirmCallback);
            rabbitTemplate.setReturnCallback(returnCallback);
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId("zheaven123456");//id + 时间戳 （必须是全局唯一的）
            rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
        }
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第五步：测试发送消息

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

package com.dwz.springboot;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import com.dwz.springboot.entity.Order;
    import com.dwz.springboot.producer.RabbitSender;
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class SpringbootTest {
        @Autowired
        private RabbitSender send;
        
        private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
        
        @Test
        public void testSender1() throws Exception {
            Map<String, Object> properties = new HashMap<String, Object>();
            properties.put("number", "12345");
            properties.put("send_time", simpleDateFormat.format(new Date()));
            send.send("Hello Rabbirmq For springboot!", properties);
            
        }
        
        @Test
        public void testSender2() throws Exception {
            Order order = new Order("001", "第一个订单");
            send.sendOrder(order);
        }
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

[回到顶部][Link 3]

## 配置消费者 ##

第一步：配置application.properties

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

spring.rabbitmq.addresses=127.0.0.1:5672
    spring.rabbitmq.username=xiaochao
    spring.rabbitmq.password=root
    spring.rabbitmq.virtual-host=/vhost_dwz
    spring.rabbitmq.connection-timeout=15000
    
    #消费端配置
    #配置手工确认模式，用于ACK的手工处理，这样我们可以保证消息的可靠性送达，或者在消费端消费失败的时候可以做到重回队列、根据业务记录日志等处理
    spring.rabbitmq.listener.simple.acknowledge-mode=manual
    #可以设置消费端的监听个数和最大个数，用于控制消费端的并发情况
    spring.rabbitmq.listener.simple.concurrency=5
    spring.rabbitmq.listener.simple.max-concurrency=10
    
    #消费端binding配置 ,以订单为例
    spring.rabbitmq.listener.order.queue.name=queue-2
    spring.rabbitmq.listener.order.queue.durable=true
    spring.rabbitmq.listener.order.exchange.name=exchange-2
    spring.rabbitmq.listener.order.exchange.durable=true
    spring.rabbitmq.listener.order.exchange.type=topic
    spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
    spring.rabbitmq.listener.order.key=springboot.*

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第二步：开发启动类Application.java

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

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

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第三步：添加主配置类

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

package com.dwz.springboot;
    
    import org.springframework.context.annotation.ComponentScan;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    @ComponentScan({"com.dwz.springboot.*"})
    public class MainConfig {
    
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

第四步：添加消息处理方法

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

package com.dwz.springboot.consumer;
    
    import java.util.Map;
    
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    
    import com.dwz.springboot.entity.Order;
    import com.rabbitmq.client.Channel;
    
    @Component
    public class RabbitReceiver {
        //此配置可以直接创建Exchange、Queue、Binding
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue(
                            name = "queue-1",
                            durable = "true"
                        ),
                exchange = @Exchange(
                            name = "exchange-1",
                            durable = "true",
                            type = "topic",
                            ignoreDeclarationExceptions = "true"
                        ),
                key = "springboot.*"
                )
        )
        @RabbitHandler
        public void onMessage(Message message, Channel  channel) throws Exception {
            System.err.println("------------------------------------");
            System.err.println("消费端Payload:" + message.getPayload());
            Long deliverytag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
            //手工ACK
            channel.basicAck(deliverytag, false);
        }
        
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue(
                            name = "${spring.rabbitmq.listener.order.queue.name}",
                            durable = "${spring.rabbitmq.listener.order.queue.durable}"
                        ),
                exchange = @Exchange(
                            name = "${spring.rabbitmq.listener.order.exchange.name}",
                            durable = "${spring.rabbitmq.listener.order.exchange.durable}",
                            type = "${spring.rabbitmq.listener.order.exchange.type}",
                            ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"
                        ),
                key = "${spring.rabbitmq.listener.order.key}"
                )
        )
    
        @RabbitHandler
        //将Message用@Payload和@Headers拆分
        //@Payload表示body里面的信息，@Headers表示properties的信息
        public void onOrderMessage(@Payload Order order, Channel  channel,
                @Headers Map<String, Object> headers) throws Exception {
            System.err.println("------------------------------------");
            System.err.println("消费端Order:" + order.getId());
            Long deliverytag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
            //手工ACK
            channel.basicAck(deliverytag, false);
        }
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

参考blog：

[https://www.cnblogs.com/zheaven/p/11915480.html][https_www.cnblogs.com_zheaven_p_11915480.html]

参考视频：

[https://www.bilibili.com/video/BV1Fz411B77A?p=34][https_www.bilibili.com_video_BV1Fz411B77A_p_34]

[https_www.cnblogs.com_guchunchao_p_13173406.html]: https://www.cnblogs.com/guchunchao/p/13173406.html
[Link 1]: https://www.cnblogs.com/guchunchao/p/13173406.html#_label0
[Link 2]: https://www.cnblogs.com/guchunchao/p/13173406.html#_label1
[961ddebeb323a10fe0623af514929fc1.png]: /images/20210527/7e8e3664e50b42cea8139dad1755de48.png
[48304ba5e6f9fe08f3fa1abda7d326ab.png]: /images/20210527/9fbceb3d269344d4b7f5186d5dcfe76f.png
[Link 3]: https://www.cnblogs.com/guchunchao/p/13173406.html#_labelTop
[https_www.cnblogs.com_zheaven_p_11915480.html]: https://www.cnblogs.com/zheaven/p/11915480.html
[https_www.bilibili.com_video_BV1Fz411B77A_p_34]: https://www.bilibili.com/video/BV1Fz411B77A?p=34

发表评论取消回复

表情：

评论列表（有 0 条评论，683人围观）

还没有评论，来说两句吧...

相关阅读

相关 SpringBoot整合RabbitMQ

目录 springboot整合rabbitmq 控制台依赖生

女爷i/ 2022年12月13日 04:29/ 0 赞/ 59 阅读

相关 RabbitMQ整合SpringBoot

RabbitMQ整合SpringBoot 一.RabbitMQ整合SpringBoot（手动ack） 1.创建SpringBoot工程 2.导入依赖

￡神魔★判官ぃ/ 2022年10月07日 02:56/ 0 赞/ 17 阅读

相关 springboot 整合 rabbitmq

pom.xml <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apa

灰太狼/ 2022年06月03日 08:55/ 0 赞/ 167 阅读

相关 SpringBoot整合RabbitMQ

一对一发送 1、配置pom包，主要是添加spring-boot-starter-amqp的支持 <dependency> <groupId>org

古城微笑少年丶/ 2022年05月24日 00:41/ 0 赞/ 275 阅读

相关 Springboot整合RabbitMQ

一：Springboot整合RabbitMQ 新建一个Module工程，工程命名为springboot-rabbitmq 在`pom.xml`中引入如下依赖内容，其中`sp

男娘i/ 2022年04月23日 08:36/ 0 赞/ 76 阅读

相关 SpringBoot整合RabbitMQ

一 RabbitMQ的介绍　　　　RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apac

红太狼/ 2022年03月02日 14:14/ 0 赞/ 366 阅读

相关 springboot整合 rabbitmq

1 添加依赖 <dependency> <groupId>org.springframework.boot</groupId> <ar

Love The Way You Lie/ 2021年12月22日 13:07/ 0 赞/ 236 阅读

相关 springboot整合RabbitMQ

使用参考网址:https://www.cnblogs.com/enjoyall/p/7767462.html?utm\_source=debugrun&utm\_medium=

电玩女神/ 2021年12月05日 05:17/ 0 赞/ 497 阅读

相关 SpringBoot--整合RabbitMQ

整合SpringBoot SpringBoot未AMQP提供了自动化配置依赖spring-boot-starter-amqp，因此首先创建项目，并添加依赖： <

ゝ一纸荒年。/ 2021年08月24日 18:10/ 0 赞/ 587 阅读

相关 RabbitMQ -springboot整合rabbitmq

课程：[https://www.cnblogs.com/guchunchao/p/13173406.html][https_www.cnblogs.com_guchunchao

Love The Way You Lie/ 2021年06月10日 20:39/ 0 赞/ 684 阅读