简单记录一下springboot+rabbitmq集成

Dear 丶 2022-05-08 15:06 157阅读 0赞

1.Windows下安装RabbitMQ需要以下几个步骤

(1)：下载erlang，原因在于RabbitMQ服务端代码是使用并发式语言erlang编写的，下载地址：http://www.erlang.org/downloads，双击.exe文件进行安装就好一路下一步下一步这里就不截图了，然后安装完成之后创建一个名为ERLANG\_HOME的环境变量，其值指向erlang的安装目录，同时将%ERLANG\_HOME%\\bin加入到Path中，最后打开命令行，输入erl，如果出现erlang的版本信息就表示erlang语言环境安装成功

(2)：下载RabbitMQ，下载地址：http://www.rabbitmq.com/，同样双击.exe进行安装就好(这里需要注意一点，默认的安装目录是C:/Program Files/....，这个目录中是存在空格符的，我们需要改变安装目录，RabbitMQ安装目录中是不允许有空格的)；

(3)：安装RabbitMQ-Plugins，这个相当于是一个管理界面，方便我们在浏览器界面查看RabbitMQ各个消息队列以及exchange的工作情况，安装方法是：打开命令行cd进入rabbitmq的sbin目录(我的目录是：E:\\software\\rabbitmq\\rabbitmq\_server-3.6.5\\sbin)，输入：rabbitmq-plugins enable rabbitmq\_management命令，稍等会会发现出现plugins安装成功的提示，默认是安装6个插件，如果你在安装插件的过程中出现了下面的错误：

![70][]

然后继续双击  rabbitmq-server.bat 启动

然后访问管理界面如下: 我这里省略掉了登陆 输入guest  guest默认账号登陆

![70 1][]

继续省略N多行rabbitmq介绍.  如果不懂可百度 查看简介

新建两个springboot 项目  一个消息生产  一个消息消费

消息生成 A

spring boot 引入相关依赖

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

消息生产者

application.properties添加一下配置  这里需要注意的是 rabbitmq 默认端口5672   ,15672是访问管理界面的端口

还有用户名密码 需要去管理界面新建 用户不然会提示无权限调用，guest 默认只能访问管理界面

spring.rabbitmq.host=192.168.1.107
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=test
    spring.rabbitmq.password=test123
    spring.rabbitmq.publisher-confirms=true
    spring.rabbitmq.publisher-returns=true
    spring.rabbitmq.template.mandatory=true

spring boot配置类，作用为指定队列，交换器类型及绑定操作

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;
    
    @Configuration
    public class RabbitConfig {
        
        //声明队列
        @Bean
        public Queue queue1() {
            return new Queue("hello.queue1", true); // true表示持久化该队列
        }
        
        @Bean
        public Queue queue2() {
            return new Queue("hello.queue2", true);
        }
        
        //声明交互器
        @Bean
        TopicExchange topicExchange() {
            return new TopicExchange("topicExchange");
        }
    
        //绑定
        @Bean
        public Binding binding1() {
            return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");
        }
        
        @Bean
        public Binding binding2() {
            return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");
        }
        
    }

共声明了2个队列，分别是hello.queue1，hello.queue2，交换器类型为TopicExchange,并与hello.queue1，hello.queue2队列分别绑定。

生产者类

import java.util.UUID;
    
    import javax.annotation.PostConstruct;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
        
        @PostConstruct
        public void init() {
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnCallback(this);
        }
    
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if (ack) {  
                System.out.println("消息发送成功:" + correlationData);  
            } else {  
                System.out.println("消息发送失败:" + cause);  
            }  
            
        }
    
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println(message.getMessageProperties().getCorrelationIdString() + " 发送失败");
            
        }
    
        //发送消息，不需要实现任何接口，供外部调用。
        public void send(String msg){
            
            CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
            
            System.out.println("开始发送消息 : " + msg.toLowerCase());
            String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();
            System.out.println("结束发送消息 : " + msg.toLowerCase());
            System.out.println("消费者响应 : " + response + " 消息处理完成");
        }
    }

要点：

1.注入RabbitTemplate

2.实现RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback接口（非必须）。  
ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。

3.实现消息发送方法。调用rabbitTemplate相应的方法即可。rabbitTemplate常用发送方法有

rabbitTemplate.send(message);   //发消息，参数类型为org.springframework.amqp.core.Message
    rabbitTemplate.convertAndSend(object); //转换并发送消息。 将参数对象转换为org.springframework.amqp.core.Message后发送
    rabbitTemplate.convertSendAndReceive(message) //转换并发送消息,且等待消息者返回响应消息。

针对业务场景选择合适的消息发送方式即可。

然后是消息消费 B

application.properties添加一下配置

spring.rabbitmq.host=192.168.1.107
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=test
    spring.rabbitmq.password=test123
    
    spring.rabbitmq.listener.concurrency=2  //最小消息监听线程数
    spring.rabbitmq.listener.max-concurrency=2 //最大消息监听线程数

消费者类

import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Receiver {
        
        @RabbitListener(queues = "hello.queue1")
        public String processMessage1(String msg) {
            System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue1队列的消息：" + msg);
            return msg.toUpperCase();
        }
        
        @RabbitListener(queues = "hello.queue2")
        public void processMessage2(String msg) {
            System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue2队列的消息：" + msg);
        }
    }

由于定义了2个队列，所以分别定义不同的监听器监听不同的队列。由于最小消息监听线程数和最大消息监听线程数都是2，所以每个监听器各有2个线程实现监听功能。

要点:

1.监听器参数类型与消息实际类型匹配。在生产者中发送的消息实际类型是String，所以这里监听器参数类型也是String。

2.如果监听器需要有响应返回给生产者，直接在监听方法中return即可。

运行测试

import java.util.Date;
    
    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.SpringJUnit4ClassRunner;
    
    import com.sam.demo.rabbitmq.Application;
    import com.sam.demo.rabbitmq.sender.Sender;
    
    @RunWith(value=SpringJUnit4ClassRunner.class)
    @SpringBootTest(classes = Application.class)
    public class RabbitTests {
        
        @Autowired
        private Sender sender;
        
        @Test
        public void sendTest() throws Exception {
            while(true){
                String msg = new Date().toString();
                sender.send(msg);
                Thread.sleep(1000);
            }
        }
    }

输出:

开始发送消息 : wed mar 29 23:20:52 cst 2017
    SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Wed Mar 29 23:20:52 CST 2017
    SimpleAsyncTaskExecutor-2 接收到来自hello.queue1队列的消息：Wed Mar 29 23:20:52 CST 2017
    结束发送消息 : wed mar 29 23:20:52 cst 2017
    消费者响应 : WED MAR 29 23:20:52 CST 2017 消息处理完成
    ------------------------------------------------
    消息发送成功:CorrelationData [id=340d14e6-cfcc-4653-9f95-29b37d50f886]
    开始发送消息 : wed mar 29 23:20:53 cst 2017
    SimpleAsyncTaskExecutor-1 接收到来自hello.queue1队列的消息：Wed Mar 29 23:20:53 CST 2017
    SimpleAsyncTaskExecutor-2 接收到来自hello.queue2队列的消息：Wed Mar 29 23:20:53 CST 2017
    结束发送消息 : wed mar 29 23:20:53 cst 2017
    消费者响应 : WED MAR 29 23:20:53 CST 2017 消息处理完成
    ------------------------------------------------
    消息发送成功:CorrelationData [id=e4e01f89-d0d4-405e-80f0-85bb20238f34]
    开始发送消息 : wed mar 29 23:20:54 cst 2017
    SimpleAsyncTaskExecutor-2 接收到来自hello.queue1队列的消息：Wed Mar 29 23:20:54 CST 2017
    SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Wed Mar 29 23:20:54 CST 2017
    结束发送消息 : wed mar 29 23:20:54 cst 2017
    消费者响应 : WED MAR 29 23:20:54 CST 2017 消息处理完成
    ------------------------------------------------

如果需要使用的其他的交换器类型，spring中都已提供实现，所有的交换器均实现org.springframework.amqp.core.AbstractExchange接口。

常用交换器类型如下：

Direct(DirectExchange)：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing\_key, 消息的routing\_key完全匹配时, 才会被交换器投送到绑定的队列中去。

Topic(TopicExchange)：按规则转发消息（最灵活）。

Headers(HeadersExchange)：设置header attribute参数类型的交换机。

Fanout(FanoutExchange)：转发消息到所有绑定队列。

![70 2][]

原本地址[https://www.jianshu.com/p/e1258c004314][https_www.jianshu.com_p_e1258c004314]

[70]: /images/20220506/6cebbd34ec3249fb8940fe49646e2531.png
[70 1]: /images/20220506/35940d37226e4f05a91872c1762e431c.png
[70 2]: /images/20220506/c9b881e9821349858c4a788dea8aefde.png
[https_www.jianshu.com_p_e1258c004314]: https://www.jianshu.com/p/e1258c004314