前言

RabbitMQ学习笔记(一)RabbitMQ部署、5种队列模型

3 SpringAMQP

3.1 介绍

AMQP(Advanced Message Queuing Protocol),是一个应用程序之间传递业务消息的标准高级消息队列协议。基于此协议的客户端与消息中间件可传递消息,并不受不同的开发语言等条件的限制。

SpringAMQP是基于AMQP协议定义的一套RabbitMQ模板,并且利用SpringBoot对其实现了自动装配,使用起来非常方便。SpringAmqp的官方地址:https://spring.io/projects/spring-amqp

SpringAMQP主要提供了三个功能:

  • 自动声明队列、交换机及其绑定关系
  • 基于注解的监听器模式,异步接收消息
  • 封装了RabbitTemplate工具,用于发送消息

3.2 简单队列模型

  • 1)引入依赖
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
  • 2)在application.yml文件中配置RabbitMQ
spring:
  rabbitmq:
    host: 192.168.153.128
    port: 5672
    virtual-host: /
    username: rabbitmq
    password: 123321
  • 3)利用RabbitTemplate实现消息发送
@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testSimpleQueue() {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, spring amqp!";
    // 发送消息
    rabbitTemplate.convertAndSend(queueName, message);
}

执行以上单元测试,在RabbitMQ管理页面查看队列中的消息:

  • 4)利用RabbitListener实现消息接收
@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息:【" + msg + "】");
    }
}
  • 5)测试结果

3.3 工作队列模型

工作队列模型即让多个消费者绑定到一个队列,共同消费队列中的消息。

  • 1)利用RabbitTemplate和循环实现消息批量发送
@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testWorkQueue() {
    // 队列名称
    String queueName = "work.queue";
    // 发送消息
    String message = "hello, work queue ";
    for (int i = 0; i < 50; i++) {
        rabbitTemplate.convertAndSend(queueName, message + i);
    }
}

执行以上单元测试,在RabbitMQ管理页面查看队列中的消息:

  • 2)利用RabbitListener实现消息接收,模拟工作队列
@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueueMessage1(String msg) {
        System.out.println("work.queue监听器1:" + msg);
    }

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueueMessage2(String msg) {
        System.out.println("work.queue监听器2:" + msg);
    }
}
  • 3)测试结果

3.4 发布/订阅模型

由上图可知,发布/订阅模型包含的角色如下:

  • Publisher:生产者,但其消息不再直接发送到队列中,而是发给exchange(交换机)。
  • Exchange:交换机。一方面,接收生产者发送的消息。另一方面,决定如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有以下3种类型:
    • Fanout:广播,将消息发送给所有绑定到交换机的队列。
    • Direct:定向,把消息发送符合指定routing key的队列。
    • Topic:通配符,把消息发送给符合routing pattern(路由模式)的队列。
  • Consumer:消费者,与其他模型一样。
  • Queue:消息队列,与其他模型一样。

需要注意的是,Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

3.4.1 Fanout广播模型

在Fanout广播模型中,Exchange(交换机)会将消息发送给所有绑定到交换机的队列。

  • 1)声明交换机和队列,及其绑定关系
@Configuration
public class FanoutConfig {

    /**
     * 声明交换机
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("star.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列1和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列2和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}
  • 2)利用RabbitTemplate发送消息到交换机
@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testFanoutExchange() {
    // 队列名称
    String exchangeName = "star.fanout";
    // 消息
    String message = "hello, fanout!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}
  • 3)利用RabbitListener接收消息
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消费者1接收到Fanout消息:【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消费者2接收到Fanout消息:【" + msg + "】");
}
  • 4)测试结果

3.4.2 Direct定向模型

在Direct定向模型中,Exchange(交换机)会把消息发送符合指定routing key的队列。

因此,交换机与队列与的绑定,不再是任意绑定,而是要指定一个RoutingKey(路由key);生产者在向Exchange发送消息时,也必须指定消息的RoutingKey

只有队列的RoutingKey与消息的RoutingKey完全一致时,才会接收到消息。

  • 1)声明交换机和队列,及其绑定关系与RoutingKey

在Fanout广播模型的案例中,使用的是@Bean的方式声明队列和交换机,比较麻烦。Spring还提供了基于注解方式来声明:

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue1"),
        exchange = @Exchange(name = "star.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者1接收到Direct的消息:【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "star.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消费者2接收到Direct的消息:【" + msg + "】");
}
  • 2)利用RabbitTemplate发送消息到交换机
@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testDirectExchange() {
    // 队列名称
    String exchangeName = "star.direct";
    // 消息
    String message = "新的风暴已经出现!";
    rabbitTemplate.convertAndSend(exchangeName, "yellow", message);
}
  • 3)测试结果

3.4.3 Topic通配符模型

Topic通配符模型和Direct定向模型一样,都是可以根据RoutingKey把消息路由到不同的队列,只不过Topic通配符模型的RoutingKey可以使用通配符!

Routingkey一般都是有一个或多个单词组成,多个单词之间以.分割,而通配符规则如下:

  • #:匹配一个或多个词
  • *:匹配恰好1个词

例如:

  • item.#:能够匹配item.insert.user 或者 item.insert

  • item.*:只能匹配item.insert

  • 1)声明交换机和队列,及其绑定关系与RoutingKey

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue1"),
        exchange = @Exchange(name = "star.topic", type = ExchangeTypes.TOPIC),
        key = {"item.#"}
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到Topic的消息:【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue2"),
        exchange = @Exchange(name = "star.topic", type = ExchangeTypes.TOPIC),
        key = {"item.*"}
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到Topic的消息:【" + msg + "】");
}
  • 2)利用RabbitTemplate发送消息到交换机
@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testTopicExchange() {
    String exchangeName = "star.topic";
    String message = "新增用户!";
    rabbitTemplate.convertAndSend(exchangeName, "item.insert.user", message);
    String message2 = "新增***!";
    rabbitTemplate.convertAndSend(exchangeName, "item.insert", message2);
}
  • 3)测试结果

3.5 消息转换器

在调用RabbitTemplate的convertAndSend()方法时,Spring会把发送的消息序列化为字节发送给MQ,接收消息时还会把字节反序列化为Java对象。

默认情况下Spring采用的序列化方式是JDK序列化。而这种方式存在下列问题:数据体积过大、有安全漏洞、可读性差。

例如执行以下单元测试:

@Test
public void testSendMap() throws InterruptedException {
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    rabbitTemplate.convertAndSend("simple.queue","", msg);
}

在RabbitMQ管理页面查看消息:

显然,可读性非常差。JDK序列化方式并不好用。而要使可读性更高,可以使用JSON方式来做序列化和反序列化。

  • 1)引入依赖
<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>
  • 2)配置消息转换器,在启动类中添加一个Bean
@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}
  • 3)再次测试

本节完,更多内容请查阅分类专栏:微服务学习笔记

感兴趣的读者还可以查阅我的另外几个专栏:

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部