零、文章前言

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

链接: pan.baidu.com/s/1lpZC6fr8… 提取码: 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>
复制代码

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

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

4. 上述三步完成后，rabbitmq的基础环境搭建完成
5. 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();
    }
}
复制代码

2. 发送消息时，使用的是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");
    }
}
复制代码

2. 监听队列
   • 监听的队列必须存在，否则将会报错
   • 监听的队列消费完成会自动确认消息
   • 如果多个队列同时监听一个队列，则消息会轮训地由不同方法处理
   • 可以在参数中指定接收类型，消息将会自动转为对应类型
   • 也可以指定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. 先启动订阅者，可以看到队列声明

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 "网络中断，请稍后再试~";
        }
    }
}
复制代码

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

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();
    }
}
复制代码

2. 声明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");
    }
}
复制代码

2. 监听队列

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();
    }
}
复制代码

2. 声明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);
    }
}
复制代码

2. 监听队列

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();
    }
}
复制代码

2. 创建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();
    }
}
复制代码

2. 队列监听

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
复制代码