消息驱动_分组消费-【官方】百战程序员_IT在线教育培训机构


xxxxxxxxxx
// 发送分组消息
    @PostMapping("/group")
    public String sendGroupMessage(@RequestParam("body") String body) {

        MyMessage myMessage = new MyMessage();
        myMessage.setPayload(body);

        // 生产消息
        streamBridge.send("myGroup-out-0", myMessage);
        return "SUCCESS";
    }

application.yml 配置


xxxxxxxxxx
server:
  port: 800
eureka:
  instance:
    # 实例名字
    instance-id: stream-producer8001
  client:
    service-url:
      # 指定单机eureka server地址
      #defaultZone: http://localhost:7001/eureka/
      defaultZone: http://localhost:7001/eureka/,http://localhost:7002/eureka/
spring:
  application:
    name: stream-producer
  rabbitmq:
    port: 5672
    host: 192.168.66.101
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        # 广播消息
        myBroadcast-out-0: # 生产者绑定名称，myBroadcast是自定义的绑定名称，out代表生产者，0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange
        # 分组消息
        myGroup-out-0: # 生产者绑定名称
          destination: my-group-topic   # 对应的真实的 RabbitMQ Exchange

消费者工程8002,8003

新增接收信息方法


xxxxxxxxxx
// 消费分组消息
    @Bean
    public Consumer<MyMessage> myGroup() {  // 方法名必须与生产消息时自定义的绑定名称一致

        return message -> {
            log.info("接收分组消息：{}", message.getPayload());
        };
    }

application.yml 配置


xxxxxxxxxx
server:
  port: 8003
eureka:
  instance:
    # 实例名字
    instance-id: stream-consumer8003
  client:
    service-url:
      # 指定单机eureka server地址
      #defaultZone: http://localhost:7001/eureka/
      defaultZone: http://localhost:7001/eureka/,http://localhost:7002/eureka/
spring:
  application:
    name: stream-consumer
  rabbitmq:
    port: 5672
    host: 192.168.66.101
    username: guest
    password: guest
  cloud:
    function:
      # 定义消费者，多个用分号分隔，当存在大于1个的消费者时，不定义不会生效
      definition: myBroadcast;myGroup
    stream:
      bindings:
        # 广播消息
        myBroadcast-in-0: # 消费者绑定名称，myBroadcast是自定义的绑定名称，in代表消费者，0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange


        # 分组消息
        myGroup-in-0: # 消费者绑定名称
          destination: my-group-topic   # 对应的真实的 RabbitMQ Exchange
          group: my-group-1   # 同一分组的消费服务，只能有一个消费者消费到消息

验证分组消息

将 Stream 工程启动两个实例，调用发送分组消息接口，查看消息消费情况。

发送分组消息接口：GEThttp://localhost:8001/group?message=hello这是一条分组消息！

自动生成的 Exchange

自动生成的 Queue

消息驱动_入门案例之消息消费者分布式请求链路追踪_为什么需要链路追踪

北京市昌平区回龙观镇南店村综合商业楼2楼226室