SpringBoot整合rabbitmq使用案例

野性酷女 2023-09-30 10:49 13阅读 0赞

### SpringBoot整合rabbitmq使用案例： ###

*   *  rabbitmq简介
     *   *  为什么使用MQ？MQ的优点
     *  1 、环境准备
     *   *  1.1添加依赖
         *  1.2 配置yml
     *  2、提供者发送消息
     *   *  2.1 send
         *  2.2 convertAndSend
     *  3、消费者接收消息
     *   *  3.1方式一
         *  3.2方式二
 *  案例源码

## rabbitmq简介 ##

### 为什么使用MQ？MQ的优点 ###

*  异步处理 ：相比于传统的串行、并行方式，提高了系统吞吐量。
 *  应用解耦 ：系统间通过消息通信，不用关心其他系统的处理。
 *  流量削锋 ：可以通过消息队列长度控制请求量；可以缓解短时间内的高并发请 求。
 *  日志处理 ：解决大量日志传输
 *  消息通讯 : 消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通 讯。比如实现点对点消息队列，或者聊天室等。

**详答 主要是：解耦、异步、削峰。**

**解耦：** A 系统发送数据到BCD三个系统，通过接口调用发送。如果 E 系统也要 这个数据呢？那如果 C 系统现在不需要了呢？A 系统负责人几乎崩溃…A 系统 跟其它各种乱七八糟的系统严重耦合，A 系统产生一条比较关键的数据，很多系 统都需要 A 系统将这个数据发送过来。如果使用 MQ，A 系统产生一条数据， 发送到 MQ 里面去，哪个系统需要数据自己去 MQ 里面消费。如果新系统需要数据，直接从 MQ 里消费即可；如果某个系统不需要这条数据了，就取消对 MQ 消息的消费即可。这样下来，A 系统压根儿不需要去考虑要给谁发送数 据，不需要维护这个代码，也不需要考虑人家是否调用成功、失败超时等情况。 就是一个系统或者一个模块，调用了多个系统或者模块，互相之间的调用很复 杂，维护起来很麻烦。但是其实这个调用是不需要直接同步调用接口的，如果用 MQ 给它异步化解耦。

**异步：** A 系统接收一个请求，需要在自己本地写库，还需要在 BCD 三个系统写 库，自己本地写库要 3ms，BCD 三个系统分别写库要 300ms、450ms、 200ms。最终请求总延时是 3 + 300 + 450 + 200 = 953ms，接近 1s，用户 感觉搞个什么东西，慢死了慢死了。用户通过浏览器发起请求。如果使用 MQ，那么 A 系统连续发送 3 条消息到 MQ 队列中，假如耗时 5ms，A 系统从 接受一个请求到返回响应给用户，总时长是 3 + 5 = 8ms。

**削峰：** 减少高峰时期对服务器压力。

## 1 、环境准备 ##

rabbitmq可以将消息放入到head中或者放入body中然后发送。

### 1.1添加依赖 ###

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

### 1.2 配置yml ###

spring:
      application:
        name: springboot_rabbit
      rabbitmq:
        host: 139.9.123.123
        virtual-host: /
        username: admin
        password: 123
        port: 5672

## 2、提供者发送消息 ##

### 2.1 send ###

@Test
        void receive() throws UnsupportedEncodingException {
        
        String message="word";
            //将消息放入头部
            final MessageProperties messageProperties = MessagePropertiesBuilder.newInstance()
                    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                    .setContentEncoding("utf-8")
                    .setHeader("hello", message)
                    .build();
    
            //将消息放入Body中
            final Message msg = MessageBuilder
                    .withBody(message.getBytes("utf-8"))
                    .andProperties(messageProperties)
                    .build();
    		//第一个参数:交换机名称  第二个参数：routingKey  第三个参数：Message
            rabbitTemplate.send("ex.boot", "key.boot", msg);
        }

### 2.2 convertAndSend ###

@Test
        void receive2() {
        
            //2、将phone 和code封装到map中
            Map<String, String> map = new HashMap<>();
            map.put("phone", "18889899090");
            map.put("code", "code");
            rabbitTemplate.convertAndSend("ex2.boot", "key2.boot", map);
        }

**说明：**

rabbitTemplate.send("ex.boot", "key.boot", msg);
    rabbitTemplate.convertAndSend("ex2.boot", "key2.boot", map);

消息发送者发送消息时，需要指定交换机名称和routingKey这样另一端消费者根据交换机名称和routingKey才能到mq中匹配到对应的队列，进行消费。

## 3、消费者接收消息 ##

### 3.1方式一 ###

RabbitConfig 配置文件

import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitConfig {
        
        /**
         * 声明交换机
         */
        @Bean
        public Exchange exchange() {
        
            //参数一：交换机名称 参数二：是否持久化 true--交换机持久化，mq重启后交换机会恢复 
            // 参数三：是否自动删除，true---使用后自动删除 参数四：参数
            return new TopicExchange("ex.boot", false, false, null);
        }
    
        /**
         * 声明队列
         */
        @Bean
        public Queue queue() {
        
            //参数一：队列名称 参数二：是否持久化 true--队列持久化，mq重启后队列会恢复
            // 参数三：是否自动删除，true---使用后自动删除 参数四：参数
            return new Queue("queue.boot", false, false, false, null);
        }
    
        /**
         * 交换机和队列绑定
         */
        @Bean
        public Binding binding() {
        
            //参数一：队列名称 参数二：交换机名称 
            // 参数三：routingKey 参数四：参数
            return new Binding("queue.boot",
                    Binding.DestinationType.QUEUE,
                    "ex.boot",
                    "key.boot",
                    null);
        }
    }

**接收方式一**：

@Component
    public class MyMessageListener1 {
        
    
        @RabbitListener(queues = "queue.boot")
        public void getMyMessage(@Payload String mes, @Header(name = "hello") String value,
                                 Message message, Channel channel) throws IOException {
        
            //value为header中数据
            System.out.println("hello = " + value);
            //mes为body中的数据
            System.out.println(mes);
    
            final long deliveryTag = message.getMessageProperties().getDeliveryTag();
            // 确认消息
            channel.basicAck(deliveryTag, false);
            // 拒收消息
    //        channel.basicReject();
        }
    }

**结果**

word
    hello = word

**接收方式二**

@Component
    public class MyMessageListener {
        
        private Integer index = 0;
        @RabbitListener(queues = "queue.boot1")
        public void getMyMessage(Message message, Channel channel) throws IOException {
        
            System.out.println(message);
    
            //从header中获取数据
            String value = message.getMessageProperties().getHeader("hello");
            //从body中获取消息
            String s = new String(message.getBody());
    
            System.out.println("hello = " + value);
            System.out.println("s = " + s);
            final long deliveryTag = message.getMessageProperties().getDeliveryTag();
            if (index % 2 == 0) {
        
                // 确认消息
                channel.basicAck(deliveryTag, false);
            } else {
        
                // 拒收消息
                channel.basicReject(deliveryTag, false);
            }
            index++;
        }

**结果**

(Body:'word' MessageProperties [headers={
        hello=word}, contentType=text/plain, contentEncoding=utf-8, contentLength=0, 
    receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=ex.boot, receivedRoutingKey=key.boot, 
    deliveryTag=1, consumerTag=amq.ctag-33BWKjwGT34_7xaetGd2NA, consumerQueue=queue.boot])
    hello = word
    s = word

### 3.2方式二 ###

@Component
    public class MessageListener {
        
    
        private Integer index = 0;
    
        @RabbitListener(bindings = @QueueBinding(
                //参数 durable exclusive autoDelete 这些默认都是false,参数可以省略
                value = @Queue(name = "queue2.boot", durable = "false", exclusive = "false", autoDelete = "false"),
                exchange = @Exchange(name = "ex2.boot", type = ExchangeTypes.TOPIC),
                key = "key2.boot"))
        public void getMyMessage(Map<String, String> msg, Channel channel, Message message) throws IOException {
        
            System.out.println("message = " + message);
            // 手机号
            String phone = msg.get("phone");
            System.out.println("phone = " + phone);
            // 验证码
            String code = msg.get("code");
            System.out.println("code = " + code);
            final long deliveryTag = message.getMessageProperties().getDeliveryTag();
            channel.basicAck(deliveryTag, false);
        }
    }

**结果**：

phone = 18889899090
    code = code

# 案例源码 #

[案例源码][Link 1]  
![在这里插入图片描述][1f21c9bd67e143f68a5285869b2fc51b.png]  
[SpringBoot整合RabbitMQ及其原理分析][SpringBoot_RabbitMQ]

[Link 1]: https://github.com/wxl52d41/rabbitmq.git
[1f21c9bd67e143f68a5285869b2fc51b.png]: https://img-blog.csdnimg.cn/1f21c9bd67e143f68a5285869b2fc51b.png
[SpringBoot_RabbitMQ]: https://blog.csdn.net/QingXu1234/article/details/130842270