SpringBoot集成RabbitMQ开发

一、基础配置开发

1.1 pom依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>${spring-boot-amqp.version}</version>
</dependency>

1.2 ymal配置

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: mitter
    password: mitter
    virtual-host: mitter_vhost # 注意：这里前面不能带/，默认的“/”理解成字符串就行，和Linux的目录斜杠还不是一回事
    publisher-confirms: true  # 消息发送到交换机确认机制，是否确认回调
    publisher-returns: true

1.3 简单队列

模拟最简单的队列，和交换机不做任何的绑定

通过注入AmqpTemplate接口的实例来实现消息的发送，AmqpTemplate接口定义了一套针对AMQP协议的基础操作。在SpringBoot中会根据配置来注入其具体实现。类似RedisTemplate等。

此时的AmqpTemplate对象其实是RabbitTemplate的实例，因为RabbitTemplate是AmqpTemplate的子类：

1604753363722.png

消息生产者：

@Service
public class AmqpServiceImpl implements IAmqpService {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void convertAndSend(String message) {
        // 队列名称：QueueConstant.QUEUE_NOTIFY_HELLO --> com.queue.notify.hello
        amqpTemplate.convertAndSend(QueueConstant.QUEUE_NOTIFY_HELLO, message);
        // rabbitTemplate是AmqpTemplate的实现类，可以用rabbitTemplate
        rabbitTemplate.convertAndSend(QueueConstant.QUEUE_NOTIFY_HELLO, "测试数据！！！！！");
    }
}

消息消费者：

使用注解@RabbitListener，配置监听的队列，可以是多个，只要有消息发送到队列，这里就可以读取到消息

@Component
public class AmqpServiceConsumer {

     private static final Logger logger = LoggerFactory.getLogger(AmqpServiceConsumer.class);

    @Autowired
    private IAmqpHelloService amqpHelloService;

    public AmqpServiceConsumer() {
    }

    @RabbitListener(queues = {QueueConstant.QUEUE_NOTIFY_HELLO})
    public void receiveSmsCodeQueue(String message) {
        logger.info("------hello：消费者处理消息------");
        logger.debug(message);
        // 方法里只是打印了下message
        amqpHelloService.receiveHelloQueue(message);
    }
}

1604753091503.png

要通过RabbitMQ发送消息的话，需要创建通道、交换机、队列，并将通道与交换机、交换机与队列绑定起来，而上述的简单例子中，为什么没看到通道、交换机的创建，也没看到绑定的操作呢？实际上在RabbitMQ中，在不创建交换机的情况下，RabbitMQ会创建一个默认的direct类型的交换机，通过RabbitMQ可视化管理界面可以看到：

1604754040012.png

1604754245465.png

Spring Boot自动配置RabbitMQ

SpringBoot整合RabbitMQ同样有个自动配置类，只不过RabbitMQ的自动配置类是由SpringBoot官方自行提供，而不像Mybatis是由Mybatis方提供的。这个自动配置类在spring-boot-autoconfigure-xxx.jar包中：

1604755010118.png

该自动配置类中自动注册了三个重要的Bean，分别是rabbitConnectionFactory、rabbitTemplate、amqpAdmin。当然RabbitMQ的配置信息由RabbitProperties类进行导入：

1604755226098.png

二、RabbitMQ交换机及工作模式

RabbitMQ的交换机Exchange有如下几种类型：

Fanout
Direct
Topic
Header

其中header类型的Exchange由于用的相对较少，所以本章主要讲述其他三种类型的Exchange。

RabbitMQ的工作模式：

发布/订阅模式：对应Fanout类型的交换机。
路由模式：对应Direct类型的交换机。
通配符模式：对应Topic类型的交换机。

2.1 发布订阅模式（Fanout）

任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。

可以理解为路由表的模式；
这种模式不需要RouteKey；
这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定；
如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。

代码示例：

FanoutConfig配置类代码，配置了两个队列和一个交换机，将两个队列都绑定到该fanout类型的交换机上：

@Configuration
public class FanoutConfig {

    public static final String FANOUT_QUEUE_NAME_1 = "fanout-queue-1";
    public static final String FANOUT_QUEUE_NAME_2 = "fanout-queue-2";
    public static final String FANOUT_EXCHANGE_NAME = "fanout-exchange";
    
    /**
     * 定义一个direct-queue-1队列
     * Queue有4个参数：
     *  1. 队列名
     *  2. durable      持久化消息队列，rabbitmq冲洗的时候，队列不会消失，不需要重建，默认true
     *  3. auto-delete  表示消息队列在没有使用的时候将被自动删除，默认false
     *  4. exclusive    表示消息队列是否只在当前的connection生效，默认false
     * @return Queue
     */
    @Bean
    public Queue fanoutQueue1() {
//        return new Queue(FANOUT_QUEUE_NAME_1);//默认情况，durable为true,exclusive为false,auto-delete为false
        return QueueBuilder.durable(FANOUT_QUEUE_NAME_1).build();
    }

    @Bean
    public Queue fanoutQueue2() {
//        return new Queue(FANOUT_QUEUE_NAME_1);//默认情况，durable为true,exclusive为false,auto-delete为false
        return QueueBuilder.durable(FANOUT_QUEUE_NAME_2).build();
    }

    @Bean
    public FanoutExchange fanoutExchange() {
//        return new FanoutExchange(FANOUT_EXCHANGE_NAME);//默认情况下，durable为true,auto-delete为false
        return (FanoutExchange) ExchangeBuilder.fanoutExchange(FANOUT_EXCHANGE_NAME).durable(true).build();
    }

    @Bean
    public Binding fanoutBinding1(FanoutExchange fanoutExchange, Queue fanoutQueue1) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    @Bean
    public Binding fanoutBinding2(FanoutExchange fanoutExchange, Queue fanoutQueue2) {
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

}

消息生产者：

@Component(value = "fanout-sender")
public class FanoutSender {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(String name) {
        String content = "hello : " + name + "，当前时间：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        amqpTemplate.convertAndSend(FanoutConfig.FANOUT_EXCHANGE_NAME, "", content);
    }

}

消息消费者：

分别监听两个队列，消息会经过fanout的交换机发送到与之绑定的两个队列上，@RabbitListener注解可以用在类上，也可以用在方法上，需配合@RabbitHandler注解一起使用，也可以直接使用在方法上了，这时就不需要@RabbitHandler注解

@Component(value = "fanout-receiver")
public class FanoutReceiver {

    /**
     * 消费者1号，监听队列1：fanout-queue-1（与交换机fanout-exchange绑定）
     * @param content 消息内容
     */
    @RabbitHandler
    @RabbitListener(queues = {FanoutConfig.FANOUT_QUEUE_NAME_1})
    public void handler1(String content) {
        System.out.println("Fanout.handler1接收到：" + content);
    }

    /**
     * 消费者2号，监听队列2：fanout-queue-2（与交换机fanout-exchange绑定）
     * @param content 消息内容
     */
    @RabbitHandler
    @RabbitListener(queues = {FanoutConfig.FANOUT_QUEUE_NAME_2})
    public void handler2(String content) {
        System.out.println("Fanout.handler2接收到：" + content);
    }

}

测试：

@RunWith(SpringRunner.class)
@SpringBootTest(classes = MqAdminApplication.class)
public class MqAdminApplicationTests {
    
    @Autowired
    private FanoutSender fanoutSender;

    @Test
    public void testFanoutExchanger() {
        fanoutSender.send("mitter");
    }

}

1604759248173.png

2.2 路由模式（Direct）

RabbitMQ默认自带Exchange，该Exchange的名字为空字符串，当然也可以自己指定名字；
在默认的Exchange下，不需要将Exchange与Queue绑定， RabbitMQ会自动绑定；而如果使用自定义的Exchange，则需要在将Exchange绑定到Queue的时候需要指定一个RouteKey；
在消息传递时需要一个RouteKey；
所有发送到Direct Exchange的消息会被转发到RouteKey中指定的Queue。
如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

代码示例：

DirectConfig配置类代码，配置两个队列，通过两个不同的routeKey绑定到同一个Exchange上：

@Configuration
public class DirectConfig {

    public static final String DIRECT_QUEUE_NAME_1 = "direct-queue-1";
    public static final String DIRECT_QUEUE_NAME_2 = "direct-queue-2";
    public static final String DIRECT_EXCHANGE_NAME = "direct-exchange";

    public static final String ROUTE_KEY_1 = "direct.route.key.1";
    public static final String ROUTE_KEY_2 = "direct.route.key.2";

    @Bean
    public Queue directQueue1() {
//        return new Queue(DIRECT_QUEUE_NAME_1);//默认情况，durable为true,exclusive为false,auto-delete为false
        return QueueBuilder.durable(DIRECT_QUEUE_NAME_1).build();
    }

    @Bean
    public Queue directQueue2() {
//        return new Queue(DIRECT_QUEUE_NAME_2);//默认情况，durable为true,exclusive为false,auto-delete为false
        return QueueBuilder.durable(DIRECT_QUEUE_NAME_2).build();
    }

    @Bean
    public DirectExchange directExchange() {
//        return new DirectExchange(DIRECT_EXCHANGE_NAME_1);//默认情况下，durable为true,auto-delete为false
        return (DirectExchange) ExchangeBuilder.directExchange(DIRECT_EXCHANGE_NAME).durable(true).build();
    }

    @Bean
    public Binding directBinding1(DirectExchange directExchange, Queue directQueue1) {
        return BindingBuilder.bind(directQueue1).to(directExchange).with(ROUTE_KEY_1);
    }

    @Bean
    public Binding directBinding2(DirectExchange directExchange, Queue directQueue2) {
        return BindingBuilder.bind(directQueue2).to(directExchange).with(ROUTE_KEY_2);
    }

}

消息生产者：

@Component("direct-sender")
public class DirectSender {

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(Integer selector) {
        String content = "hello，我是%d号，当前时间：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        String routeKey = "";
        if (selector == 1) {
            content = String.format(content, 1);
            routeKey = DirectConfig.ROUTE_KEY_1;
        } else if (selector == 2) {
            content = String.format(content, 2);
            routeKey = DirectConfig.ROUTE_KEY_2;
        } else {
            content = String.format(content, 3);
            // 如果通过route key没有发现与该交换机绑定的队列，则抛弃此消息
            routeKey = "direct.route.key.3";
        }
        amqpTemplate.convertAndSend(DirectConfig.DIRECT_EXCHANGE_NAME, routeKey, content);
    }

}

消息消费者：

@Component(value = "direct-receiver")
public class DirectReceiver {

    @RabbitHandler
    @RabbitListener(queues = {DirectConfig.DIRECT_QUEUE_NAME_1})
    public void directHandler1(String content) {
        System.out.println("Direct.directHandler1接收到：" + content);
    }

    @RabbitHandler
    @RabbitListener(queues = {DirectConfig.DIRECT_QUEUE_NAME_2})
    public void directHandler2(String content) {
        System.out.println("Direct.directHandler2接收到：" + content);
    }

}

测试：

@RunWith(SpringRunner.class)
@SpringBootTest(classes = MqAdminApplication.class)
public class MqAdminApplicationTests {
    
    @Autowired
    private DirectSender directSender;

    @Test
    public void testDirectExchanger() {
        // 通过routing key发送到队列direct-queue-1上
        directSender.send(1);
        // 通过routing key发送到队列direct-queue-2上
        directSender.send(2);
        // 如果通过route key没有发现与该交换机绑定的队列，则抛弃此消息
        directSender.send(3);
    }

}

注意：direct类型的消息，只要被一个消费者消费掉后就不能再被另外一个消费者消费了

1604761751131.png

2.3 通配符模式（Topic）

任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定的Queue上。

这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。
这种模式需要RouteKey，也需要提前绑定Exchange与Queue。
在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
“#”表示0个或若干个关键字，“*****”表示一个关键字。如“log.*”能与“log.warn”匹配，但是无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。
同样，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。

代码示例：

TopicConfig配置类，声明了两个队列，分别对应两个routeKey:topic.#和topic.*

@Configuration
public class TopicConfig {

    public static final String TOPIC_QUEUE_NAME_1 = "topic-queue-1";
    public static final String TOPIC_QUEUE_NAME_2 = "topic-queue-2";
    public static final String TOPIC_EXCHANGE_NAME = "topic-exchange";

    public static final String ROUTE_KEY_1 = "topic.#";
    public static final String ROUTE_KEY_2 = "topic.*";

    @Bean
    public TopicExchange topicExchange() {
//        return new TopicExchange(TOPIC_EXCHANGE_NAME);//默认情况下，durable为true,auto-delete为false
        return (TopicExchange) ExchangeBuilder.topicExchange(TOPIC_EXCHANGE_NAME).durable(true).build();
    }
    
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE_NAME_1);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE_NAME_2);
    }

    @Bean
    public Binding topicBinding1(TopicExchange topicExchange, Queue topicQueue1) {
        return BindingBuilder.bind(topicQueue1).to(topicExchange).with(ROUTE_KEY_1);
    }

    @Bean
    public Binding topicBinding2(TopicExchange topicExchange, Queue topicQueue2) {
        return BindingBuilder.bind(topicQueue2).to(topicExchange).with(ROUTE_KEY_2);
    }
}

消息生产者：

@Component(value = "topic-sender")
public class TopicSender {

    private static final String TOPIC_PREFIX = "topic.";

    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(String selector) {
        String content = "hello，当前时间：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        amqpTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, TOPIC_PREFIX + selector, content);
    }

}

消息消费者：

@Component(value = "topic-receiver")
public class TopicReceiver {

    @RabbitHandler
    @RabbitListener(queues = {TopicConfig.TOPIC_QUEUE_NAME_1})
    public void handler1(String content) {
        System.out.println("Topic.handler1接收到：" + content);
    }

    @RabbitHandler
    @RabbitListener(queues = {TopicConfig.TOPIC_QUEUE_NAME_2})
    public void handler2(String content) {
        System.out.println("Topic.handler2接收到：" + content);
    }

}

测试：

@RunWith(SpringRunner.class)
@SpringBootTest(classes = MqAdminApplication.class)
public class MqAdminApplicationTests {
    
    @Autowired
    private TopicSender topicSender;

    @Test
    public void testTopicExchanger() {
        // 匹配topic.#和topic.*
        topicSender.send("message");
        // 只能匹配topic.#
        topicSender.send("message.a");
    }

}

1604763187500.png

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,214评论 6赞 481
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,307评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 152,543评论 0赞 341
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,221评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,224评论 5赞 371
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,007评论 1赞 284
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,313评论 3赞 399
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,956评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,441评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,925评论 2赞 323
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,018评论 1赞 333
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,685评论 4赞 322
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,234评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,240评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,464评论 1赞 261
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,467评论 2赞 352
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,762评论 2赞 345

SpringBoot集成RabbitMQ开发

一、基础配置开发

1.1 pom依赖

1.2 ymal配置

1.3 简单队列

消息生产者：

消息消费者：

Spring Boot自动配置RabbitMQ

二、RabbitMQ交换机及工作模式

2.1 发布订阅模式（Fanout）

2.2 路由模式（Direct）

2.3 通配符模式（Topic）

推荐阅读更多精彩内容