SpringBoot-RabbitMQ篇(1)-四大交换器

矫情吗；* 2022-10-15 14:57 138阅读 0赞

### 文章目录 ###

*  零、文章前言
 *  一、环境搭建
 *  二、四大交换器
 *   *  2.1 direct - 直连交换器
     *   *  2.1.1 消息发送者
         *  2.2.2 消息接收者
         *  2.1.3 消息发布订阅
     *  2.2 topic - 主题交换器
     *   *  2.2.1 消息发送者
         *  2.2.2 消息接收者
         *  2.2.3 消息发布订阅
     *  2.3 fanout - 广播交换器
     *   *  2.3.1 消息发送者
         *  2.32 消息接收者
         *  2.3.3 消息发布订阅
     *  2.4 headers - 头交换器
     *   *  2.4.1 消息发送者
         *  2.4.2 消息接收者
         *  2.4.3 消息发布订阅

# 零、文章前言 #

1.  SpringBoot-RabbitMQ高级篇系列开始更新，本系列主要为SpringBoot整合RabbitMQ，实现高可用、可靠传递等
2.  核心有四大交换器、死信队列、可靠传递、异常消费处理
3.  系列共计三篇文章，本系列核心主讲整合和企业级内容，需要先具备SpringBoot和RabbitMQ基础知识
4.  文章源码放到了网盘，没有放git仓库，需要的自行下载，脚本等信息在common下面
5.  个人水平有限，有错误的地方欢迎指正  
    **链接**: [https://pan.baidu.com/s/1lpZC6fr8f4WFV58Zfo\_FVw][https_pan.baidu.com_s_1lpZC6fr8f4WFV58Zfo_FVw] **提取码**: qtvi

# 一、环境搭建 #

1.  采用maven多module模式，共计创建三个子module
    
     *  common：通用实体信息
     *  rabbitmq-publisher：消息发布者，基于SpringBoot
     *  rabbitmq-subscriber：消息订阅者，基于SpringBoot
2.  在消息发布者和订阅者两个项目中加入rabbitmq maven依赖

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

1.  在两个项目中加入rabbitmq的配置信息

spring:
      rabbitmq:
        host: xxx.xxx.xxx.xxx
        port: 5672
        username: username
        password: password
        # 虚拟主机，需要后台先配置
        # virtual-host: springboot

1.  上述三步完成后，rabbitmq的基础环境搭建完成
2.  rabbitmq配置属性类
    
     *  org.springframework.boot.autoconfigure.amqp.RabbitProperties

# 二、四大交换器 #

## 2.1 direct - 直连交换器 ##

### 2.1.1 消息发送者 ###

1.  在消息发布者中新建配置类，声明交换器信息
    
     *  只用声明交换器，队列和交换器绑定是订阅者操作
     *  不同的类型提供不同的交换器
     *  如果只声明交换器并不会创建交换器，而是**绑定**时或者**发送消息**时才创建

import org.springframework.amqp.core.DirectExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class AmqpPublisherConfig { 
        @Bean
        public DirectExchange emailDirectExchange() { 
            // 声明方式一
            // return new DirectExchange("exchange.direct.springboot.email");
            // 声明方式二
            return ExchangeBuilder.directExchange("exchange.direct.springboot.email").build();
        }
    }

1.  发送消息时，使用的是RabbitTemplate，为SpringBoot提供的RabbitMQ消息发送器
    
     *  org.springframework.amqp.rabbit.core.RabbitTemplate
     *  发送消息示例

import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import javax.annotation.Resource;
    
    @RestController
    public class PublishController { 
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        @RequestMapping("/direct")
        public Object direct(String message) { 
            try { 
                rabbitTemplate.convertAndSend("交换器", "路由键", message);
                return message;
            } catch (AmqpException e) { 
                System.out.println(e.getMessage());
                return "网络中断，请稍后再试~";
            }
        }
    }

### 2.2.2 消息接收者 ###

1.  接收者需要配置以下内容
    
     *  交换器：直接new对应的交换器类型
     *  队列：只有Queue类型，通过名称区分
     *  交换器和队列的绑定：通过BindingBuilder.bind(队列).to(交换器).with(路由键);
     *  只声明交换器和队列绑定，并不会马上创建，而是在发送消息或者监听队列时才会创建

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class AmqpSubscriberConfig { 
       /** * 直连交换器 */
        @Bean
        public DirectExchange emailDirectExchange() { 
            // 声明方式一
            // return new DirectExchange("exchange.direct.springboot.email");
            // 声明方式二
            return ExchangeBuilder.directExchange("exchange.direct.springboot.email").build();
        }
    
        /** * 声明队列 */
        @Bean
        public Queue emailQueue() { 
            // 声明方式一
            // return new Queue("queue.direct.springboot.email");
            // 声明方式二
            return QueueBuilder.durable("queue.direct.springboot.email").build();
        }
    
        /** * 交换器和队列绑定 */
        @Bean
        @Resource
        public Binding emailBiding(Queue emailQueue, DirectExchange emailDirectExchange) { 
            // 将路由使用路由键绑定到交换器上
            return BindingBuilder.bind(emailQueue).to(emailDirectExchange).with("springboot.email.routing.key");
        }
    }

1.  监听队列
    
     *  监听的队列必须存在，否则将会报错
     *  监听的队列消费完成会自动确认消息
     *  如果多个队列同时监听一个队列，则消息会轮训地由不同方法处理
     *  可以在参数中指定接收类型，消息将会自动转为对应类型
     *  也可以指定Message参数获取对应消息信息
        
         *  org.springframework.amqp.core.Message
         *  获取消息属性：message.getMessageProperties()
         *  获取消息内容：message.getBody()

import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /** * 消息订阅监听 */
    @Component
    public class SubscriberListener { 
        /** * direct监听，相同监听队列消息将会轮流处理 */
        @RabbitListener(queues = "queue.direct.springboot.email")
        public void receiver01(String msg) { 
            System.out.println("receiver01 message = " + msg);
        }
    
        @RabbitListener(queues = "queue.direct.springboot.email")
        public void receiver02(String msg) { 
            System.out.println("receiver02 message = " + msg);
        }
    }

### 2.1.3 消息发布订阅 ###

1.  先启动订阅者，可以看到队列声明

![4f7d9805297dcdc4c65dfe5485948ee8.png_height_446_id_E03TA_originHeight_446_originWidth_1155_originalType_binary_size_0_status_done_style_none_width_1155][]  
2. 启动发布者，然后发布消息

*  http://127.0.0.1:8071/direct?message=direct

import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import javax.annotation.Resource;
    
    @RestController
    public class PublishController { 
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        @RequestMapping("/direct")
        public Object direct(String message) { 
            try { 
                // 指定发送的交换器和路由键
                rabbitTemplate.convertAndSend("exchange.direct.springboot.email", "springboot.email.routing.key", message);
                return message;
            } catch (AmqpException e) { 
                System.out.println(e.getMessage());
                return "网络中断，请稍后再试~";
            }
        }
    }

1.  订阅者会轮流收到信息

receiver01 message = direct
    receiver02  message = direct
    receiver01 message = direct
    receiver02  message = direct
    receiver01 message = direct
    receiver02  message = direct

## 2.2 topic - 主题交换器 ##

### 2.2.1 消息发送者 ###

1.  声明topic交换器

import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class BlogPublisherConfig { 
        @Bean
        public Exchange blogTopicExchange() { 
            return ExchangeBuilder.topicExchange("exchange.topic.springboot.blog").build();
        }
    }

1.  声明controller

@RequestMapping("/topic")
    public Object topic(String routingKey, String message) { 
        rabbitTemplate.convertAndSend("exchange.topic.springboot.blog", routingKey, message);
        return routingKey + " : " + message;
    }

### 2.2.2 消息接收者 ###

1.  声明交换器、三个队列、队列的绑定
    
     *  \*：匹配一个串
     *  \#：匹配一个或者多个串

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;
    
    import javax.annotation.Resource;
    
    @Configuration
    public class BlogSubscriberConfig { 
        /** * 主题交换器 */
        @Bean
        public TopicExchange blogTopicExchange() { 
            return ExchangeBuilder.topicExchange("exchange.topic.springboot.blog").build();
        }
    
        @Bean
        public Queue blogJavaQueue() { 
            return QueueBuilder.durable("queue.topic.springboot.blog.java").build();
        }
    
        @Bean
        public Queue blogMqQueue() { 
            return QueueBuilder.durable("queue.topic.springboot.blog.mq").build();
        }
    
        @Bean
        public Queue blogAllQueue() { 
            return QueueBuilder.durable("queue.topic.springboot.blog.all").build();
        }
    
        @Bean
        @Resource
        public Binding blogJavaBinding(TopicExchange blogTopicExchange, Queue blogJavaQueue) { 
            return BindingBuilder.bind(blogJavaQueue).to(blogTopicExchange).with("springboot.blog.java.routing.key");
        }
    
        @Bean
        @Resource
        public Binding blogMqBinding(TopicExchange blogTopicExchange, Queue blogMqQueue) { 
            return BindingBuilder.bind(blogMqQueue).to(blogTopicExchange).with("springboot.blog.mq.routing.key");
        }
    
        @Bean
        @Resource
        public Binding blogAllBinding(TopicExchange blogTopicExchange, Queue blogAllQueue) { 
            // #: 匹配一个或者多个 *：匹配一个
            return BindingBuilder.bind(blogAllQueue).to(blogTopicExchange).with("springboot.blog.#.routing.key");
        }
    }

1.  监听队列

import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    @Service
    public class BlogService { 
        /** * topic监听 */
        @RabbitListener(queues = "queue.topic.springboot.blog.java")
        public void blogJavaListener(String message) { 
            System.out.println("blogJavaListener message = " + message);
        }
    
        @RabbitListener(queues = "queue.topic.springboot.blog.mq")
        public void blogMqListener(String message) { 
            System.out.println("blogMqListener message = " + message);
        }
    
        @RabbitListener(queues = "queue.topic.springboot.blog.all")
        public void blogAllaListener(String message) { 
            System.out.println("blogAllListener message = " + message);
        }
    }

### 2.2.3 消息发布订阅 ###

1.  发布者发送消息
    
     *  http://localhost:8071/topic?routingKey=springboot.blog.java.routing.key&message=hello
     *  http://localhost:8071/topic?routingKey=springboot.blog.mq.routing.key&message=hello
2.  订阅者收到消息
    
     *  全匹配和模糊匹配
     *  全匹配无论是哪个都会被匹配上

blogJavaListener message = hello
    blogAllListener message = hello
        
    blogAllListener message = hello
    blogMqListener message = hello

## 2.3 fanout - 广播交换器 ##

### 2.3.1 消息发送者 ###

1.  声明fanout交换器

import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class NoticePublisherConfig { 
        @Bean
        public Exchange radioFanoutExchange() { 
            return ExchangeBuilder.fanoutExchange("exchange.fanout.springboot.radio").build();
        }
    }

1.  声明controller

@RequestMapping("/fanout")
    public Object fanout(String message) { 
        rabbitTemplate.convertAndSend("exchange.fanout.springboot.radio", null, message);
        return message;
    }

### 2.32 消息接收者 ###

1.  创建交换器、路由键、绑定
    
     *  不需要使用路由键

import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.Resource;
    
    @Configuration
    public class NoticeSubscriberConfig { 
        @Bean
        public FanoutExchange radioFanoutExchange() { 
            return ExchangeBuilder.fanoutExchange("exchange.fanout.springboot.radio").build();
        }
    
        @Bean
        public Queue radioQueue() { 
            return QueueBuilder.durable("queue.fanout.springboot.radio").build();
        }
    
        @Bean
        @Resource
        public Binding radioBinding(FanoutExchange radioFanoutExchange, Queue radioQueue) { 
            // 广播交换器绑定没有路由键，只要绑定即可收到
            return BindingBuilder.bind(radioQueue).to(radioFanoutExchange);
        }
    }

1.  监听队列

import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    @Service
    public class NoticeService { 
    
        @RabbitListener(queues = "queue.fanout.springboot.radio")
        public void radioListener(String message) { 
            System.out.println("radioListener message = " + message);
        }
    }

### 2.3.3 消息发布订阅 ###

1.  发布者发送消息
    
     *  http://localhost:8071/fanout?message=fanout
2.  订阅者收到消息

radioListener message = fanout

## 2.4 headers - 头交换器 ##

### 2.4.1 消息发送者 ###

1.  headers模式通过头匹配，会忽略路由键
2.  发送者需要创建队列

import org.springframework.amqp.core.HeadersExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class HeadersPublisherConfig { 
        @Bean
        public Exchange radioHeadersExchange() { 
            return ExchangeBuilder.headersExchange("exchange.headers.springboot.headers").build();
        }
    }

1.  创建controller发送消息
    
     *  MessageProperties和Message包是：org.springframework.amqp.core
     *  需要创建MessageProperties对象用于设置头信息
     *  Message用于存储消息和消息属性信息

@RequestMapping("/headers")
    public Object headers(@RequestParam Map<String, String> param) { 
        MessageProperties properties = new MessageProperties();
        properties.setHeader("name", param.get("name"));
        properties.setHeader("token", param.get("token"));
        Message mqMessage = new Message(param.get("message").getBytes(), properties);
        rabbitTemplate.convertAndSend("exchange.headers.springboot.headers", null, mqMessage);
        return properties;
    }

### 2.4.2 消息接收者 ###

1.  接收者和上面三种一样，同样需要声明交换器、队列、绑定

*  在队列绑定时需要使用不同规则
    
     *  BindingBuilder.bind(headersQueue01).to(headersExchange).whereAll(key).match()
        
         *  所有字段属性和值全部匹配
     *  BindingBuilder.bind(headersQueue02).to(headersExchange).whereAny(key).match()
        
         *  任意字段属性和值全部匹配
     *  BindingBuilder.bind(headersQueue03).to(headersExchange).whereAll(“name”, “token”).exist()
        
         *  指定所有属性字段存在
     *  BindingBuilder.bind(headersQueue03).to(headersExchange).whereAny(“name”, “token”).exist()
        
         *  指定任意属性存在
 *  headerMap中存放的属性就是发送者中封装的属性，属性完全匹配则正确路由到此处

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.HeadersExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.Resource;
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class HeadersSubscriberConfig { 
        @Bean
        public HeadersExchange headersExchange() { 
            return ExchangeBuilder.headersExchange("exchange.headers.springboot.headers").build();
        }
    
        @Bean
        public Queue headersQueue01() { 
            return QueueBuilder.durable("queue.headers.springboot.01").build();
        }
    
        @Bean
        public Queue headersQueue02() { 
            return QueueBuilder.durable("queue.headers.springboot.02").build();
        }
    
        @Bean
        public Queue headersQueue03() { 
            return QueueBuilder.durable("queue.headers.springboot.03").build();
        }
    
        @Bean
        @Resource
        public Binding headers01Binding(HeadersExchange headersExchange,Queue headersQueue01) { 
            Map<String, Object> key = new HashMap<>(4);
            key.put("name", "java");
            key.put("token", "001");
            return BindingBuilder.bind(headersQueue01).to(headersExchange).whereAll(key).match();
        }
    
        @Bean
        @Resource
        public Binding headers02Binding(HeadersExchange headersExchange,Queue headersQueue02) { 
            Map<String, Object> key = new HashMap<>(4);
            key.put("name", "java");
            key.put("token", "002");
            return BindingBuilder.bind(headersQueue02).to(headersExchange).whereAny(key).match();
        }
    
        @Bean
        @Resource
        public Binding headers03Binding(HeadersExchange headersExchange,Queue headersQueue03) { 
            // name和token都需要存在
            return BindingBuilder.bind(headersQueue03).to(headersExchange).whereAll("name", "token").exist();
            // 任意name或者token存在
            // return BindingBuilder.bind(headersQueue03).to(headersExchange).whereAny("name", "token").exist();
        }
    }

1.  队列监听

import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    @Service
    public class HeadersService { 
    
        @RabbitListener(queues = "queue.headers.springboot.01")
        public void headers01Listener(String message) { 
            System.out.println("headers01Listener message = " + message);
        }
    
        @RabbitListener(queues = "queue.headers.springboot.02")
        public void headers02Listener(String message) { 
            System.out.println("headers02Listener message = " + message);
        }
    
        @RabbitListener(queues = "queue.headers.springboot.03")
        public void headers03Listener(String message) { 
            System.out.println("headers03Listener message = " + message);
        }
    }

### 2.4.3 消息发布订阅 ###

1.  发送消息
    
     *  http://localhost:8071/headers?name=java&token=001&message=headers
     *  http://localhost:8071/headers?name=java&token=002&message=headers
     *  http://localhost:8071/headers?name=mq&token=003&message=headers
2.  接收消息

headers01Listener message = headers
    headers02Listener message = headers
    headers03Listener message = headers
        
    headers02Listener message = headers
    headers03Listener message = headers
        
    headers03Listener message = headers

[https_pan.baidu.com_s_1lpZC6fr8f4WFV58Zfo_FVw]: https://pan.baidu.com/s/1lpZC6fr8f4WFV58Zfo_FVw
[4f7d9805297dcdc4c65dfe5485948ee8.png_height_446_id_E03TA_originHeight_446_originWidth_1155_originalType_binary_size_0_status_done_style_none_width_1155]: /images/20221014/e4ffa65f917a4d8d966c3443fd484a23.png