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

本文主要聊SpringBoot整合RabbitMQ,主要分为生产者和消费者两个工程,目录结构如下:

2D035572-C360-5A72-C380-A29FAB0AE1A1.png

先简单说一下RabbitMQ的一些核心概念:

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的默认配置就是它们:

16351C93-62ED-A0B0-73B8-99A09DAB391F.png

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

(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注解,指定了接收消息的队列。

示例代码仓库:GitHub - qiuxinfa/rabbitmq-study

收藏 (0)
评论列表
正在载入评论列表...
我是有底线的
为您推荐
    暂时没有数据