美文网首页
SpringCloud 2020.0.4 系列之 Stream

SpringCloud 2020.0.4 系列之 Stream

作者: 追风人聊Java | 来源:发表于2022-01-14 23:07 被阅读0次

1. 概述

老话说的好:事情太多,做不过来,就先把事情记在本子上,然后理清思路、排好优先级,一件一件的去完成。

言归正传,今天我们来聊一下 SpringCloud 的 Stream 组件,SpringCloud Stream 是一个构建消息驱动微服务的框架,简单理解就是在 SpringCloud 中操作消息中间件的组件。

这里我们以 RabbitMQ 为例,介绍一下如何使用 Stream 实现 消息广播 和 消息分组。

RabbitMQ 镜像模式集群的搭建,可参见我的另一篇文章《RabbitMQ 3.9.7 镜像模式集群的搭建》(https://www.jianshu.com/p/187325419e95

在早期的 SpringCloud 版本中常使用 @Input、@Output、@EnableBinding 和 @StreamListener 注解开发生产者与消费者。

官方原文:Deprecated as of 3.1 in favor of functional programming model。

SpringCloud 2020.0.4 版本中,已经不推荐这么开发了,因此这里我们也使用新的写法(函数式编程方式) 开发。

闲话不多说,直接上代码。

2. 消息广播

2.1 主要依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- 健康检查 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>

2.2 消息实体类

@Setter
@Getter
public class MyMessage implements java.io.Serializable {

    // 消息体
    private String payload;
}

2.3 生产者

    @Autowired
    private StreamBridge streamBridge;

    // 发送广播消息
    @PostMapping("/broadcast")
    public String sendBroadcastMessage(@RequestParam("body") String body) {

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

        // 生产消息
        // 第一个参数是绑定名称,格式为:自定义的绑定名称-out-0,myBroadcast是自定义的绑定名称,out代表生产者,0是固定写法
        // 自定义的绑定名称必须与消费方法的方法名保持一致
        // 第二个参数是发送的消息实体
        streamBridge.send("myBroadcast-out-0", myMessage);
        return "SUCCESS";
    }

2.4 消费者

@Slf4j
@Component
// 消费者
public class Concumer {

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

        return message -> {
            log.info("接收广播消息:{}", message.getPayload());
        };
    }

2.5 application.yml 配置

spring:
  application:
    name: my-stream-new
  rabbitmq:   # RabbitMQ 配置
    addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000
  cloud:
    function:
      # 定义消费者,多个用分号分隔,当存在大于1个的消费者时,不定义不会生效
      definition: myBroadcast
    stream:
      bindings:
        
        # 广播消息
        myBroadcast-in-0:   # 消费者绑定名称,myBroadcast是自定义的绑定名称,in代表消费者,0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange
        myBroadcast-out-0:  # 生产者绑定名称,myBroadcast是自定义的绑定名称,out代表生产者,0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange

注意:spring.cloud.function.definition 属性非常重要,务必配置

2.6 验证广播消息

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

发送广播消息接口:

POST http://localhost:49000/stream/broadcast?body=这是一条广播消息!

自动生成的 Exchange

image

自动生成的 Queue

image

两个实例的消费情况

image image

3. 消息分组

消息分组后,同一分组的消费服务,只能有一个消费者消费到消息。

3.1 生产者

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

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

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

3.2 消费者

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

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

3.3 application.yml 配置

spring:
  application:
    name: my-stream-new
  rabbitmq:   # RabbitMQ 配置
    addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000
  cloud:
    function:
      # 定义消费者,多个用分号分隔,当存在大于1个的消费者时,不定义不会生效
      definition: myBroadcast;myGroup
    stream:
      bindings:        
        # 广播消息
        myBroadcast-in-0:   # 消费者绑定名称,myBroadcast是自定义的绑定名称,in代表消费者,0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange
        myBroadcast-out-0:  # 生产者绑定名称,myBroadcast是自定义的绑定名称,out代表生产者,0是固定写法
          destination: my-broadcast-topic   # 对应的真实的 RabbitMQ Exchange

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

3.4 验证分组消息

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

发送分组消息接口:

POST http://localhost:49000/stream/group?body=这是一条分组消息!

自动生成的 Exchange

image

自动生成的 Queue

image

两个实例的消费情况

image image

4. 综述

今天聊了一下 SpringCloud Stream 组件中 消息广播 与 消息分组 的实现 ,希望可以对大家的工作有所帮助。

欢迎帮忙点赞、评论、转发、加关注 :)

关注追风人聊Java,每天更新Java干货。

相关文章

  • SpringCloud 2020.0.4 系列之 Stream

    1. 概述 老话说的好:出错不怕,怕的是出了错,却不去改正。如果屡次出错,无法改对,就先记下了,然后找援军解决。 ...

  • SpringCloud 2020.0.4 系列之 Stream

    1. 概述 老话说的好:事情太多,做不过来,就先把事情记在本子上,然后理清思路、排好优先级,一件一件的去完成。 言...

  • SpringCloud 2020.0.4 系列之 Stream

    1. 概述 老话说的好:对待工作要有责任心,不仅要完成自己的部分,还要定期了解整体的进展。 言归正传,我们在开发产...

  • SpringCloud 2020.0.4 系列之Eureka

    1. 概述 老话说的好:遇见困难,首先要做的是积极的想解决办法,而不是先去泄气、抱怨或生气。 言归正传,微服务是当...

  • SpringCloud 2020.0.4 系列之 Bus

    1. 概述 老话说的好:会休息的人才更会工作,身体是革命的本钱,身体垮了,就无法再工作了。 言归正传,之前我们聊了...

  • SpringCloud 2020.0.4 系列之 Gateway

    1. 概述 老话说的好:做人要有幽默感,懂得幽默的人才会活的更开心。 言归正传,今天我们来聊聊 SpringClo...

  • SpringCloud 2020.0.4 系列之 Config

    1. 概述 老话说的好:一条路走不通,就去走另一条路,A计划执行不下去,就按B计划执行,多准备几套方案总是有用的。...

  • SpringCloud 2020.0.4 系列之 Sleuth

    1. 概述 老话说的好:安全不能带来财富,但盲目的冒险也是不可取的,大胆筹划,小心实施才是上策。 言归正传,微服务...

  • SpringCloud 2020.0.4 系列之 Feign

    1. 概述 老话说的好:任何问题都有不止一种的解决方法,当前的问题没有解决,只是还没有发现解决方法,而并不是无解。...

  • SpringCloud 2020.0.4 系列之Hystrix看

    1. 概述 老话说的好:沉默是金,有时适当的沉默,比滔滔不绝更加有效。 言归正传,前面我们聊了有关 Hystrix...

网友评论

      本文标题:SpringCloud 2020.0.4 系列之 Stream

      本文链接:https://www.haomeiwen.com/subject/jiwzcrtx.html