一、认识 RabbitMQ

RabbitMQ 简介以 AMQP 协议:

(1)RabbitMQ 是开源的消息代理和队列服务器,用来通过普通协议在完全不同的应用之间共享数据,RabbitMQ 底层是用了 Erlang 语言来编写的,并且 RabbitMQ 是基于 AMQP 协议的.
(2)RabbitMQ 不仅仅可以使用 java 客户端进行编写,且可以使用其他的语言(python,php等…),它提供了丰富的API

RabbitMQ 的优点:

(1)开源,性能优秀,稳定性保障
(2)与 SpringAMQP 完美的整合,API 丰富 (Spring基于 RabbitMQ 提供了一套框架,叫做AMQP 框架)这套框架不仅提供了原生的 RabbitMQ,而且还提供了丰富可扩张的API帮助开发人员更好的去应用
(3)集群模式丰富,表达式配置,HA模式,镜像队列模型
说明:(保证数据不丢失的提前做到高可靠性,可用性)普遍使用的镜像队列模式
(4)AMQP 全称:Advanced Message Queuing Protocl AMQP 翻译过来:高级消息队列协议

二、安装 RabbitMQ

这里为了方便在 docker 上安装 RabbitMQ
(1)首先搜索 RabbitMQ 的安装命令:https://hub.docker.com/_/rabbitmq

 docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3-management
1


安装成功:

对于映射的端口需要注意两个:15672 和 5672,其中 15672 是管理的端口,5672 是通讯的端口


可以看到docker容器 15672 对应的主机映射端口是 32771,


默认的用户名和密码都是 guest

三、SpringBoot 整合 RabbitMQ

1. 加入依赖

2. 配置 application.properties

注意这里的端口选择 32781,对应rabbitmq 容器的端口是 5672,选择通讯端口,切忌不用选择管理端口

spring.rabbitmq.host=192.168.245.133
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=32781
1234

3. 直接交换模式

直接交换模式可以参考:https://blog.csdn.net/fakerswe/article/details/81508963

所谓“直接连接交换机”就是:Producer(生产者)投递的消息被DirectExchange (交换机)转发到通过routingkey绑定到具体的某个Queue(队列),把消息放入队列,然后Consumer从Queue中订阅消息

这里的具体某个队列是由参数 routingKey 控制的,消息是通过这个参数往队列上发的

RabbitMQ消息模型的核心思想(core idea): 生产者会把消息发送给RabbitMQ的交换中心(Exchange),Exchange的一侧是生产者,另一侧则是一个或多个队列,由Exchange决定一条消息的生命周期–发送给某些队列,或者直接丢弃掉。

(1)配置 RabbitDirectConfig

@Configuration
public class RabbitDirectConfig {

    public final static String DIRECTNAME = "yolo-direct";

    // 消息队列
    @Bean
    Queue queue() {
        return new Queue("hello.yolo");
    }

    @Bean
    DirectExchange directExchange() {
        return new DirectExchange(DIRECTNAME, true, false);
    }
    // 将 queue 和 directExchange 绑定到一起
    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
    }
}
123456789101112131415161718192021

(2)消费者

@Component
public class DirectReceiver {
    // 监听队列
    @RabbitListener(queues = "hello.yolo")
    public void handler1(String msg) {
        System.out.println("handler1>>>" + msg);
    }
}
12345678

(3)测试:生产者

 	@Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void contextLoads() {
        //将消息转发到 routingKey 为 hello.yolo 的队列,对应 DirectReceiver 的监听队列名
        rabbitTemplate.convertAndSend("hello.yolo", "hello yolo! ni hao!");
    }
1234567

4. 广播模式

可以参考:https://blog.csdn.net/fakerswe/article/details/81455340

简单的讲,就是把交换机(Exchange)里的消息发送给所有绑定该交换机的队列,忽略routingKey。


由图可知,生产者把消息发送到交换机后,由交换机发送给消费者队列。消费者队列如果想要接收到交换机里的消息,那么需要保证:队列绑定的交换机名称要和交换机一致,这个是广播模式的关键,也是MQ后续所有模式最粗略的前提。

这里消息是通过生产者发往交换机的,然后交换机再发送给绑定的队列
(1)配置广播模式

@Configuration
public class RabbitFanoutConfig {
    public static final String FANOUTNAME = "yolo-fanout";

    /**
     * 队列1
     * @return
     */
    @Bean
    Queue queueOne() {
        return new Queue("queue-one");
    }

    /**
     * 队列2
     * @return
     */
    @Bean
    Queue queueTwo() {
        return new Queue("queue-two");
    }

    /**
     * 交换机
     * @return
     */
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUTNAME, true, false);
    }

    /**
     * 绑定队列1
     * @return
     */
    @Bean
    org.springframework.amqp.core.Binding bindingOne() {
        return BindingBuilder.bind(queueOne()).to(fanoutExchange());
    }

    /**
     * 绑定队列2
     * @return
     */
    @Bean
    Binding bindingTwo() {
        return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
    }
}
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849

(2)消费者

/**
 * 定义接收器:消费者
 */
@Component
public class FanoutReceiver {
    /**
     * 接收消息队列1
     * @param msg
     */
    @RabbitListener(queues = "queue-one")
    public void handler1(String msg) {
        System.out.println("FanoutReceiver:handler1:" + msg);
    }
    /**
     * 接收消息队列2
     * @param msg
     */
    @RabbitListener(queues = "queue-two")
    public void handler2(String msg) {
        System.out.println("FanoutReceiver:handler2:" + msg);
    }
}
12345678910111213141516171819202122

(3)测试:生产者

  /**
     * 往交换机上发送信息
     */
    @Test
    public void test1() {
        rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME,null,"hello fanout!");
    }
1234567

这里需要注意,需要先启动消费者,再启动生产者,否则先启动生产者,exchange接收到消息后发现没有队列对其感兴趣,就会将消息丢掉,此时跟 routingKey 无关

队列1和队列2 均收到了消息

5. 主题路由匹配模式

可参考:https://blog.csdn.net/weixin_43770545/article/details/90902788

假如你想在淘宝上买一双运动鞋,那么你是不是会在搜索框中搜“XXX运动鞋”,这个时候系统将会模糊匹配的所有符合要求的运动鞋,然后展示给你。
所谓“主题路由匹配交换机”也是这样一个道理,但是使用时也有一定的规则。

String routingkey = “testTopic.#”;
String routingkey = “testTopic.*”;
12

* 表示只匹配一个词
#表示匹配多个词

(1)配置 topic 模式

@Configuration
public class RabbitTopicConfig {
    public static final String TOPICNAME = "yolo-topic";

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(TOPICNAME, true, false);
    }

    @Bean
    Queue xiaomi() {
        return new Queue("xiaomi");
    }

    @Bean
    Queue huawei() {
        return new Queue("huawei");
    }

    @Bean
    Queue phone() {
        return new Queue("phone");
    }

    @Bean
    Binding xiaomiBinding() {
        //xiaomi.# 表示如果路由的 routingKey 是以xiaomi 开头就会发送到 xiaomi 这个队列上
        return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
    }

    @Bean
    Binding huaweiBinding() {
        //huawei.#
        return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
    }
    @Bean
    Binding phoneBinding() {
        // #.phone.# 表示routingKey 中包含 phone 就会被发送到 phone 这个队列上
        return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
    }
}

123456789101112131415161718192021222324252627282930313233343536373839404142

(2) 消费者

@Component
public class TopicReceiver {
    @RabbitListener(queues = "xiaomi")
    public void handler1(String msg) {
        System.out.println("TopicReceiver:handler1:" + msg);
    }

    @RabbitListener(queues = "huawei")
    public void handler2(String msg) {
        System.out.println("TopicReceiver:handler2:" + msg);
    }

    @RabbitListener(queues = "phone")
    public void handler3(String msg) {
        System.out.println("TopicReceiver:handler3:" + msg);
    }
}
1234567891011121314151617

(3)测试:生产者

  @Test
    public void test2() {
        //可以被小米的队列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "xiaomi.news", "小米新闻");
        //可以被手机的队列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "vivo.phone", "vivo 手机");
        //可以被华为和手机的队列收到
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "huawei.phone", "华为手机");
    }
123456789

6. header 模式

这种模式使用的是 header 中的 key/value (键值对) 匹配队列,也和 routingKey 无关

(1)配置 config

@Configuration
public class RabbitHeaderConfig {
    public static final String HEADERNAME = "yolo-header";

    @Bean
    HeadersExchange headersExchange() {
        return new HeadersExchange(HEADERNAME, true, false);
    }

    @Bean
    Queue queueName() {
        return new Queue("name-queue");
    }

    @Bean
    Queue queueAge() {
        return new Queue("age-queue");
    }

    @Bean
    Binding bindingName() {
        Map<String, Object> map = new HashMap<>();
        //
        map.put("name", "yolo");
        //whereAny 表示消息的header中只要有一个header匹配上map中的key,value,就把消息发送到对应的队列上
        return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();
    }

    @Bean
    Binding bindingAge() {
        //只要有,age 这个字段,就发送到相应的队列上去
        return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();
    }
}

1234567891011121314151617181920212223242526272829303132333435

(2)消费者

@Component
public class HeaderReceiver {
    @RabbitListener(queues = "name-queue")
    public void handler1(byte[] msg) {
        System.out.println("HeaderReceiver:handler1:" + new String(msg, 0, msg.length));
    }

    @RabbitListener(queues = "age-queue")
    public void handler2(byte[] msg) {
        System.out.println("HeaderReceiver:handler2:" + new String(msg, 0, msg.length));
    }
}
123456789101112

(3)测试

  @Test
    public void test3() {
        //对应 RabbitHeaderConfig 中的map 的 key / value
        Message nameMsg = MessageBuilder.withBody("hello yolo !".getBytes()).setHeader("name","yolo").build();
        Message ageMsg = MessageBuilder.withBody("hello 99 !".getBytes()).setHeader("age","99").build();
        //此时发送的消息接收,跟 routingKey无关,跟消息的 header 内容有关
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);
    }
123456789


如果更改,header 里的键值对:则无法匹配成功,队列收不到该信息

Logo

CSDN联合极客时间,共同打造面向开发者的精品内容学习社区,助力成长!

更多推荐