一篇文章教你将JAVA的RabbitMQz与SpringBoot整合

更新时间：2021年09月16日 09:39:24 作者：qiuxinfa123

这篇文章主要介绍了如何将JAVA的RabbitMQz与SpringBoot整合，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

1.虚拟主机vhost：vhost是物理隔离的，你可以将vhost看作是一个个小型的RabbitMQ

2.交换机exchange：生产者发送的消息不是直接到达队列的，而是交换机，然后交换机再根据路由key，路由到指定的队列，可以理解为一个分发消息的角色

3.队列：存放消息的地方，消费者拿到的消息就是通过队列，可以理解为存放消息的容器

4.bindingKey和routingKey：这两个可能比较容易搞混，bindingKey是交换机和队列建立关系时候的key，而routingKey则是生产者发送消息时候的key。在DirectExchange类型的交换机中，bindingKey和routingKey是一致的，而在TopicExchange类型的交换机中，两者一般是不一致的。

开发环境：RabbitMQ:3.7.7、SpringBoot2.1.7

引入相关依赖：

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>
        <!-- rabbitmq依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>
    </dependencies>

在生产者和消费者工程的application.yml文件添加RabbitMQ配置：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

如果是和上面一样的值，其实也可以不用配置，因为RabbitMQ的默认配置就是它们：

接下来，看看几种交换机的使用场景，包括：

（1）fanout：发布订阅型，可以理解为广播

（2）direct：直连型，可以理解为点对点

（3）topic：通配符模式

一、fanout：发布订阅型

在生产者工程，添加如下配置：

@Configuration
public class FanoutRabbitmqConfig {
    public static final String EMAIL_QUEUE = "email_fanout";
    public static final String SMS_QUEUE = "sms_fanout";
    public static final String EXCHANGE_NAME = "exchange_fanout";
    // 发布订阅模式，不用routingKey
    @Bean
    public FanoutExchange getFanoutExchange(){
        return new FanoutExchange(EXCHANGE_NAME);
    }
    @Bean
    public Queue getFanoutEmailQueue(){
        return new Queue(EMAIL_QUEUE);
    }
    @Bean
    public Queue getFanoutSmsQueue(){
        return new Queue(SMS_QUEUE);
    }
    // 绑定交换机和队列
    @Bean
    public Binding fanoutEmailExchange(){
        return BindingBuilder.bind(getFanoutEmailQueue()).to(getFanoutExchange());
    }
    @Bean
    public Binding fanoutSmsExchange(){
        return BindingBuilder.bind(getFanoutSmsQueue()).to(getFanoutExchange());
    }
    // 用于测试找到了交换机，但是没有找到队列的情况
    @Bean
    public FanoutExchange notBindingQueue(){
        return new FanoutExchange("testNotBindingQueue");
    }
}

可以看到声明了名为exchange_fanout的交换机和两个队列，并进行了绑定，如果向exchange_fanout交换机发送消息，则两个队列都会收到消息。

二、direct：直连型

在生产者工程添加如下配置：

@Configuration
public class RoutingRabbitmqConfig {
    public static final String EMAIL_QUEUE = "email_routing";
    public static final String SMS_QUEUE = "sms_routing";
    public static final String EXCHANGE_NAME = "exchange_routing";
    public static final String EMAIL_ROUTINGKEY = "inform.email.routingKey";
    public static final String SMS_ROUTINGKEY = "inform.sms.routingKey";
    @Bean
    public DirectExchange getDirectExchange(){
        return new DirectExchange(EXCHANGE_NAME);
    }
    @Bean
    public Queue getDirectEmailQueue(){
        return new Queue(EMAIL_QUEUE);
    }
    @Bean
    public Queue getDirectSmsQueue(){
        return new Queue(SMS_QUEUE);
    }
    // 绑定交换机和队列
    @Bean
    public Binding directEmailExchange(){
        return BindingBuilder.bind(getDirectEmailQueue()).to(getDirectExchange()).with(EMAIL_ROUTINGKEY);
    }
    @Bean
    public Binding directSmsExchange(){
        return BindingBuilder.bind(getDirectSmsQueue()).to(getDirectExchange()).with(SMS_ROUTINGKEY);
    }
}

和fanout类型不同的是，这里使用了routingKey进行交换机和队列的绑定，如果发送消息时的routingKey是inform.sms.routingKey，那么只有sms_routing这个队列可以收到消息，email_routing队列则不会收到消息，这正是直连的含义所在。

三、topic：通配符模式

topic类型的交换机，常使用的通配符是：*和#

（1）星号*：匹配一个词

（2）井号：匹配0个或多个词

词是通过点分割的，被点分割的就是一个词

先看配置：

@Configuration
public class TopicsRabbitmqConfig {
    public static final String EMAIL_QUEUE = "email_topics";
    public static final String SMS_QUEUE = "sms_topics";
    public static final String EXCHANGE_NAME = "exchange_topics";
    public static final String EMAIL_ROUTINGKEY = "inform.#.email.#";
    // * 必须匹配一个词，不能是0个
//    public static final String EMAIL_ROUTINGKEY = "inform.*.email";
    public static final String SMS_ROUTINGKEY = "inform.#.sms.#";
    @Bean
    public TopicExchange getTopicsExchange(){
        return new TopicExchange(EXCHANGE_NAME);
    }
    @Bean
    public Queue getTopicsEmailQueue(){
        return new Queue(EMAIL_QUEUE);
    }
    @Bean
    public Queue getTopicsSmsQueue(){
        return new Queue(SMS_QUEUE);
    }
    // 绑定交换机和队列
    @Bean
    public Binding topicsEmailExchange(){
        return BindingBuilder.bind(getTopicsEmailQueue()).to(getTopicsExchange()).with(EMAIL_ROUTINGKEY);
    }
    @Bean
    public Binding topicsSmsExchange(){
        return BindingBuilder.bind(getTopicsSmsQueue()).to(getTopicsExchange()).with(SMS_ROUTINGKEY);
    }
}

和direct不同的是：bindingKey中可以含有星号和井号，可以看到：队列email_topics的bindingKey（注意，这是bindingKey，不是routingKey）是inform.#.email.#，而队列sms_topics则绑定了inform.#.sms.#。如果发送消息的routingKey是inform开头，后面含有email，则email_topics队列会收到消息，而sms_topics队列不会收到消息。

topic类型交换机的强大之处在于，它可以替代前面两种交换机：

（1）如果bindingKey只有一个井号，在是fanout

（2）如果bindingKey不含有星号和井号，则是direct

四、消费者端接收消息

消费者接收消息配置如下：

@Component
public class TopicsConsumer {
    // topics模式
    @RabbitListener(queues = TopicsRabbitmqConfig.EMAIL_QUEUE)
    public void receiveEmail(String msg){
        System.out.println("topics收到email消息： "+msg);
    }
    @RabbitListener(queues = TopicsRabbitmqConfig.SMS_QUEUE)
    public void receiveSms(String msg){
        System.out.println("topics收到sms消息： "+msg);
    }
    // routing模式
    @RabbitListener(queues = RoutingRabbitmqConfig.EMAIL_QUEUE)
    public void receiveRoutingEmail(String msg){
        System.out.println("routing收到email消息： "+msg);
    }
    @RabbitListener(queues = RoutingRabbitmqConfig.SMS_QUEUE)
    public void receiveRoutingSms(String msg){
        System.out.println("routing收到sms消息： "+msg);
    }
    // 发布订阅模式
    @RabbitListener(queues = FanoutRabbitmqConfig.EMAIL_QUEUE)
    public void receiveFanoutEmail(String msg){
        System.out.println("fanout收到email消息： "+msg);
    }
    @RabbitListener(queues = FanoutRabbitmqConfig.SMS_QUEUE)
    public void receiveFanoutSms(String msg){
        System.out.println("fanout收到sms消息： "+msg);
    }
}

这里主要是使用了RabbitListener注解，指定了接收消息的队列。

总结

本片文章就到这里了，希望能够给你带来帮助，也希望您能够多多关注脚本之家的更多内容！

您可能感兴趣的文章:

简单了解java等待唤醒机制原理及使用
这篇文章主要介绍了简单了解java等待唤醒机制原理及使用,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-12-12
如何解决线程太多导致java socket连接池出现的问题
这篇文章主要介绍了如何解决线程太多导致socket连接池出现的问题,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-12-12
Java中消息队列任务的平滑关闭详解
对于消息队列的监听，我们一般使用Java写一个独立的程序，在Linux服务器上运行。程序启动后，通过消息队列客户端接收消息，放入一个线程池进行异步处理，并发的快速处理。这篇文章主要给大家介绍了关于Java中消息队列任务的平滑关闭的相关资料，需要的朋友可以参考下。
2017-11-11
java使用SFTP上传文件到资源服务器
这篇文章主要介绍了java使用SFTP上传文件到资源服务器，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2018-12-12
Java多线程高并发中的Fork/Join框架机制详解
本文主要介绍了 Java 多线程高并发中的 Fork/Join 框架的基本原理和其使用的工作窃取算法(work-stealing)、设计方式和部分实现源码，感兴趣的朋友跟随小编一起看看吧
2021-11-11
java开发AOP基础JdkDynamicAopProxy
这篇文章主要为大家介绍了java开发AOP基础JdkDynamicAopProxy源码示例解析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2023-07-07
springmvc字符编码过滤器CharacterEncodingFilter的使用
这篇文章主要介绍了springmvc字符编码过滤器CharacterEncodingFilter的使用，具有很好的参考价值，希望对大家有所帮助。
2021-08-08
Java中 shuffle 算法的使用
本篇文章，小编将为大家介绍，在Java中 shuffle 算法的使用，有需要的朋友可以参考一下
2013-04-04
mybatis 多表关联mapper文件写法操作
这篇文章主要介绍了mybatis 多表关联mapper文件写法操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-12-12
Mybatis Mybatis-Plus传入多个参数的处理方式
这篇文章主要介绍了Mybatis Mybatis-Plus传入多个参数的处理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-05-05