Spring Cloud Stream3.2.1消费者分组与消息分区

Spring Cloud Stream3.2.1消费者分组与消息分区 1.前言

在入门实践中，我们提到过启动应用程序创建的是临时队列，在应用停止时会被自动删除，其原因是当我们未指定group属性时，创建的都是会自动删除的匿名队列。当我们指定了group属性后，队列名称则为destination.group。例如sms-exchange.group01

2.队列持久化 2.1 yml配置

spring:
  rabbitmq:
    host: 192.168.1.92
    port: 5672
    username: stone
    password: 123456
    virtual-host: /test
  cloud:
    stream:
      bindings:
        sms-out-0: # 自定义生产者通道名称
          destination: sms-exchange # 自定义Exchange交换机名称，生产者和消费者需要配置同一个Exchange
        sms-in-0: # 自定义消费者通道名称
          destination: sms-exchange # 自定义Exchange交换机名称，生产者和消费者需要配置同一个Exchange
          group: group01

启动应用我们会得到一个持久化的队列

当生产者向指定Exchange发送了一条消息时，处于同一个group的消费者只会消费一次消息，不会出现重复消费，当然如果消费者处于两个不同的group，消费消息是不受影响的。

3.消息分区

RabbitMQ本身不支持消息分区，有时候，将数据发送到特定的分区是非常有必要的。我们可以将一些特定的消息发送到同一个分区。

3.1 yml配置

spring:
  application:
    name: sms
  rabbitmq:
    host: 192.168.1.2
    port: 5672
    username: guest
    password: guest
    virtual-host: /test
  cloud:
    stream:
      bindings:
        sms-out-0: # 自定义生产者通道名称
          destination: sms-exchange # 自定义Exchange交换机名称，生产者和消费者需要配置同一个Exchange
          producer:
            partitionKeyexpression: payload #payload的含义是让算法根据我们消息的内容进行分区计算，我们可以自定义分区算法
            partition-count: 2
        sms-in-0: # 自定义消费者通道名称
          destination: sms-exchange # 自定义Exchange交换机名称，生产者和消费者需要配置同一个Exchange
          group: group01
          consumer:
            partitioned: true # 消费开启分区
      instance-count: 2 # 应用程序部署的实例数
      instance-index: 0 # 应用程序实例的索引数，如果instance-count=2的话，那instance-index分别就是0,1

这时候我们在启动两个消费者实例，就会发现接收到的消息是进过分区处理的

最后，欢迎关注微信公众号一起交流