Stream基础篇-Stream入门应用

向右看齐 2023-01-19 07:19 30阅读 0赞

总括：

SpringCloud   Stream  提供了中间件消费端的代理功能  
在消费端代理的同时也对队列进行了分区代理，任何消息类型的队列都会分区，可以指定特征分区和指定分区消费

分区：  
可以在配置文件配置：发送客户端配置

客户端代理：  
@EnableBinding(MySink.class)  
@StreamListener(MySink.INPUT)

配置消息中间中间配置方式可以使用常规的配置---input，output代替的chanhenl就是管道没有对应的队列用管道名前缀，  
此管道的前缀就是channel名字后缀是一串随机数  
也可以使用stream的配置方式配置，用此配置方式可以配置常规配置，和bing，input管道和队列的对应关系，分区代理，  
使用是直接用不用重新写消费端

注解用于拦截识别，拦截到之后做相应的处理---客户端代理实现原理  
接口定义变量（管道-主题信息），有注解的抽象方法，stream识别到这个之后生成客户端代理

public interface MySinkSender \{  
        @Output(MySink.CHANNEL)  ---------有这个注解即可反向代理客户端，指定分区发送也可再次  
        MessageChannel output();  
    \}

package com.cloud.shf.stream.sink;  
public interface MySink \{  
    String CHANNEL = "myInput";  
    @Input(MySink.CHANNEL)------有这个注解已经可以反向代理客户端，  
    SubscribableChannel input();  
\}

@StreamListener(MySink.CHANNEL)----是一种等效的额接收客户端代理  
    public void myReceive(Object payload) \{  
        LOGGER.info("Received from \{\} channel : \{\}", MySink.CHANNEL, payload.toString());  
    \}

实例：

Spring Cloud Stream是什么?

Spring Cloud Stream是一个用来为微服务应用构建消息驱动能力的框架。它可以基于Spring Boot来创建独立的、可用于生产的Spring应用程序。它通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动的微服务应用。简单的说，Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration，实现了一套轻量级的消息驱动的微服务框架。通过使用Spring Cloud Stream，可以有效地简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。由于Spring Cloud Stream基于Spring Boot实现，所以它秉承了Spring Boot的优点，实现了自动化配置的功能帮忙我们可以快速的上手使用，其主要包含以下核心概念和内容：  
 Spring Cloud Stream的应用模型  
 绑定抽象  
 持久化发布／订阅支持  
 消费组  
 消息分区  
 可插拔绑定api 应用模型

应用模型

应用通过Spring Cloud Stream插入的input和output通道与外界交流。通道通过指定中间件的Binder实现与外部代理连接。

绑定抽象

Binder绑定器是Spring Cloud Stream中一个非常重要的概念。在没有绑定器这个概念的情况下，我们的Spring Boot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性，这使得我们实现的消息交互逻辑就会非常笨重，因为对具体的中间件实现细节有太重的依赖，当中间件有较大的变动升级、或是更换中间件的时候，我们就需要付出非常大的代价来实施。

通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道，使得应用程序不需要再考虑各种不同的消息中间件实现。当我们需要升级消息中间件，或是更换其他消息中间件产品时，我们要做的就是更换它们对应的Binder绑定器而不需要修改任何Spring Boot的应用逻辑。这一点在上一章实现消息总线时，从RabbitMQ切换到Kafka的过程中，已经能够让我们体验到这一好处。

目前版本的Spring Cloud Stream为主流的消息中间件产品RabbitMQ和Kafka提供了默认的Binder实现。

持久化发布／订阅支持

应用间通信遵照发布-订阅模型，消息通过共享主题进行广播。下图所示，显示了交互的Spring Cloud Stream 应用的典型布局。

在Spring Cloud Stream中的消息通信方式遵循了发布-订阅模式，当一条消息被投递到消息中间件之后，它会通过共享的Topic主题进行广播，消息消费者在订阅的主题中收到它并触发自身的业务逻辑处理。这里所提到的Topic主题是Spring Cloud Stream中的一个抽象概念，用来代表发布共享消息给消费者的地方。在不同的消息中间件中，Topic可能对应着不同的概念，比如：在RabbitMQ中的它对应了Exchange、而在Kakfa中则对应了Kafka中的Topic。

消费组

由于发布－订阅模型使得共享主题的应用之间连接更简便，创建给定应用的不同实例来进行弹性扩张的能力也同样重要。如果存在多个应用实例，那么同一应用的不同实例便会成为相互竞争的消费者，其中应该只有一个实例处理给定消息。

Spring Cloud Stream通过消费组的概念给这种情况进行建模。每一个单独的消费者可以使用spring.cloud.stream.bindings.input.group属性来指定一个组名字。下图中展示的消费者们，这一属性被设置为spring.cloud.stream.bindings.input.group=hdfsWrite或者spring.cloud.stream.bindings.input.group=average。

所有订阅给定目标的组都会收到发布消息的一个拷贝，但是  
 每一个组内只有一个成员  
 会收到该消息。默认情况下，当我们没有为应用指定消费组的时候，Spring Cloud Stream会为其分配一个独立的匿名消费组。所以，如果同一主题下所有的应用都没有指定消费组的时候，当有消息被发布之后，所有的应用都会对其进行消费，因为它们各自都属于一个独立的组中。 一般来说，将应用绑定到给定目标的时候，最好指定一个消费，并且每一个通道均需要独立设定一个消费组。 这样可以防止应用实例收到重复的消息。（除非存在重复收到的需求，如刷新所有实例的配置）。

持久性

与Spring Cloud Stream中的可选应用模型一样，消费者组订阅是持久的。也就是说，一个绑定的实现确保组的订阅者是持久的，一旦组中至少有一个成员创建了订阅，这个组就会收到消息，即使组中所有的应用都被停止了，组仍然会收到消息。

注：自然情况下，  
 匿名订阅者是非持久化的  
 。对于某些绑定实现（如rabbitmq），可以创建非持久化（non－durable）组订阅。

消息分区

Spring Cloud Stream支持在给定应用程序的多个实例之间对数据进行分区。在分区方案中，每个topic均可以被构建为多个分区。一个或者多个生产者应用实例给多个消费者应用实例发送消息并确保  
 相同特征的数据被同一消费者实例  
 处理。

Spring Cloud Stream为分区提供了通用的抽象实现，用来在消息中间件的上层实现分区处理，所以它对于消息中间件自身是否实现了消息分区并不关心，这使得Spring Cloud Stream为不具备分区功能的消息中间件也增加了分区功能扩展。（如具备分区特性的kafka或者不带分区的特性的rabbitmq）。

分区是有状态处理中的一个关键概念，无论是性能还是一致性的原因，分区都是至关重要的，  
 当生产者将消息数据发送给多个消费者实例时，保证拥有共同特征的消息数据始终是由同一个消费者实例接收和处理  
 。。例如，在时间窗平均计算示例中，来自任何给定传感器的所有测量值都由相同应用程序实例处理是很重要的。

注：要设置分区处理方案，您必须配置数据生成和数据消耗两端。

本章概要

1、工程结构说明；

2、构建消费者工程实例；

3、自定义Channel；

4、生产者另一种实现；

工程结构说明

1、主要构建如下两个模块：

其中receiver也就是我们常说的消费者，sender自然就是生产者。。本小节主要例子将集中在receiver工程实现。

2、在父  
 pom.xml中添加如下主要依赖：

<properties>  
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>  
    <java.version>1.8</java.version>  
</properties>  
   
<parent>  
    <groupId>org.springframework.boot</groupId>  
    <artifactId>spring-boot-starter-parent</artifactId>  
    <version>1.5.6.RELEASE</version>  
</parent>  
   
<dependencies>  
   
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-starter-test</artifactId>  
        <scope>test</scope>  
    </dependency>  
   
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-starter-logging</artifactId>  
    </dependency>  
   
</dependencies>  
   
<dependencyManagement>  
    <dependencies>  
        <dependency>  
            <groupId>org.springframework.cloud</groupId>  
            <artifactId>spring-cloud-dependencies</artifactId>  
            <version>Dalston.SR5</version>  
            <type>pom</type>  
            <scope>import</scope>  
        </dependency>  
    </dependencies>  
</dependencyManagement>

构建消费者工程实例

stream-receiver工程实现如下：

1、在  
 pom.xml中添加如下依赖：

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

其等价使用  
 spring-cloud-stream-binder-rabbit依赖。

2、新建一个启动类，作为一个普通的springboot项目：

package com.cloud.shf.stream;  
@SpringBootApplication  
public class ReceiverApp \{  
    public static void main(String\[\] args) \{  
        SpringApplication.run(ReceiverApp.class, args);  
    \}  
\}

3、SpringCloudStream已经预定义了Sink、Source、Processor，具体如下

\{  
 @link   
 org.springframework.cloud.stream.messaging.Processor\},

\{  
 @link   
 org.springframework.cloud.stream.messaging.Source\},

\{  
 @link   
 org.springframework.cloud.stream.messaging.Sink\}

采用Sink作为默认的消息订阅通道，定义如下：

package com.cloud.shf.stream.sink;  
@EnableBinding(value = \{Sink.class\})  
public class SinkReceiver \{  
    @StreamListener(Sink.INPUT)  
    public void receive(Object payload) \{  
        LOGGER.info("Received from default channel : \{\}", payload.toString());  
    \}  
\}

Note：  
 将@EnableBinding注解至spring应用的一个配置类中，即可将spring应用变成Spring Cloud Stream应用。@EnableBinding注解本身就包含@Configuration注解，并且会触发Spring Cloud Stream基本配置；  
 将Sink.class作为@EnableBinding注解的参数，其指定了需要绑定的目标接口；  
 @StreamListener注解中描述具体监听的通道名称；

4、在application.properties文件中添加如下rabbitmq配置：

\#configure rabbitmq

spring.rabbitmq.host  
 =  
 localhost

spring.rabbitmq.port  
 =  
 5672

spring.rabbitmq.username  
 =  
 soul

spring.rabbitmq.password  
 =  
 123456

5、通过单元测试，实现消息的生产者：

package com.cloud.shf.stream;  
@RunWith(SpringRunner.class)  
@EnableBinding(value = \{ReceiverAppTest.SinkSender.class\})  
public class ReceiverAppTest \{  
    @Autowired  
    private SinkSender sinkSender;  
   
    @Test  
    public void sinkSenderTester() \{  
        sinkSender.output().send(MessageBuilder.withPayload("produce a message to " + Sink.INPUT + " channel").build());  
    \}  
   
    public interface SinkSender \{  
        @Output(Sink.INPUT)  
        MessageChannel output();  
    \}  
\}

Note：  
 如上生产者定义主要参考org.springframework.cloud.stream.messaging.Source定义：

public interface Source \{  
   String OUTPUT = "output";  
   @Output(Source.OUTPUT)  
   MessageChannel output();  
\}

为了与接收消息的通道一致，故修改@Output注解参数为Sink.INPUT；

6、启动服务，可以看到如下log，声明了一个queue，并添加了订阅者：

7、查看rabbitmq控制台的Queus页签中存在上述队列：

8、并执行单元测试可以看到如下log信息：

小节，以上即实现了一个最简单的demo示例。

自定义Channel

上一小节采用默认预定义的Sink已经实现了消息的消费，并模拟Source定义了消息的生产者，同理，本小节将参考源代码SInk设计，实现自定义Channel的消息处理。

1、首先简单看下Sink的定义：

public interface Sink \{  
   String INPUT = "input";  
   @Input(Sink.INPUT)  
   SubscribableChannel input();  
\}

Note：  
 其通道名称定义为input；  
 @Input注解表示输入通道，当前应用将接收消息；对应的@Output注解表示输出通道，当前应用将发送消息；注解参数表示通道的名称，如果不提供名称，则当前注解方法名即作为通道名称；  
 必须以SubscribableChannel作为返回类型；

2、参考Sink的定义，自定义如下，通道名称为  
 myInput：

package com.cloud.shf.stream.sink;  
public interface MySink \{  
    String CHANNEL = "myInput";  
    @Input(MySink.CHANNEL)  
    SubscribableChannel input();  
\}

3、在  
 @EnableBinding注解中加入当前定义的  
 MySink接口：

package com.cloud.shf.stream.sink;  
@EnableBinding(value = \{Sink.class, MySink.class\})  
public class SinkReceiver \{  
    @StreamListener(MySink.CHANNEL)  
    public void myReceive(Object payload) \{  
        LOGGER.info("Received from \{\} channel : \{\}", MySink.CHANNEL, payload.toString());  
    \}  
\}

Note：  
 必须在@EnableBinding注解加入新定义的消费消息接口，否则无法被注册；

4、添加单元测试如下：

package com.cloud.shf.stream;  
@RunWith(SpringRunner.class)  
@EnableBinding(value = \{ReceiverAppTest.SinkSender.class,  
        ReceiverAppTest.MySinkSender.class\})  
public class ReceiverAppTest \{  
    @Autowired  
    private MySinkSender mySinkSender;  
    @Test  
    public void mySinkSenderTester() \{  
        mySinkSender.output().send(MessageBuilder.withPayload("produce a message to " + MySink.CHANNEL + " channel").build());  
    \}  
    public interface MySinkSender \{  
        @Output(MySink.CHANNEL)  
        MessageChannel output();  
    \}  
\}

Note：  
 必须在@EnableBinding注解加入新定义的生产消息接口，否则无法被注册；  
 返回类型必须为MessageChannel；

5、启动服务，可以看到如下log：

6、队列中新增消息队列：

7、执行单元测试后，可以看到接收到如下消息：

小节：以上即实现了自定义消息通道，比较简单。。

生产者另一种实现

上述两个小节中的生产者均通过直接注入定义的生产接口获取  
 MessageChannel实例，然后发送消息，其实也可以直接注入  
 MessageChannel实例来完成消息的发送，本小节将实现模拟。

1、在  
 MySink继续添加如下两个通道定义：

package com.cloud.shf.stream.sink;  
public interface MySink \{  
    String OUTPUT1\_CHANNEL="OutPut-1";  
    String OUTPUT2\_CHANNEL="OutPut-2";  
   
    @Input(OUTPUT1\_CHANNEL)  
    SubscribableChannel input1();  
   
    @Input(OUTPUT2\_CHANNEL)  
    SubscribableChannel input2();  
\}

2、在  
 SinkReceiver添加如下对两个通道的监听实现：

@StreamListener(MySink.OUTPUT1\_CHANNEL)  
public void myReceive1(Object payload) \{  
    LOGGER.info("Received from \{\} channel : \{\}", MySink.OUTPUT1\_CHANNEL, payload.toString());  
\}  
   
@StreamListener(MySink.OUTPUT2\_CHANNEL)  
public void myReceive2(Object payload) \{  
    LOGGER.info("Received from \{\} channel : \{\}", MySink.OUTPUT2\_CHANNEL, payload.toString());  
\}

3、在单元测试中继续添加新增通道生产逻辑：

@RunWith(SpringRunner.class)  
@EnableBinding(value = \{ReceiverAppTest.SinkSender.class,  
        ReceiverAppTest.MySinkSender.class,  
        ReceiverAppTest.MyOutputSource.class\})  
public class ReceiverAppTest \{  
    @Resource(name = MySink.OUTPUT1\_CHANNEL)  
    private MessageChannel send1;  
   
    @Resource(name = MySink.OUTPUT2\_CHANNEL)  
    private MessageChannel send2;  
   
    @Test  
    public void myOutputSourceTester() \{  
        send1.send(MessageBuilder.withPayload("produce a message to " + MySink.OUTPUT1\_CHANNEL + " channel").build());  
        send2.send(MessageBuilder.withPayload("produce a message to " + MySink.OUTPUT2\_CHANNEL + " channel").build());  
    \}  
   
    public interface MyOutputSource \{  
        @Output(MySink.OUTPUT1\_CHANNEL)  
        MessageChannel output1();  
   
        @Output(MySink.OUTPUT2\_CHANNEL)  
        MessageChannel output2();  
    \}  
\}

Note：  
 直接注入MessageChannel实例，并采用通道名称；

4、执行单元测试：

小节：  
 通过注入生产接口然后获取  
 MessageChannel实例，其实与直接注入  
 MessageChannel是一样的，特别注意直接注入  
 MessageChannel时，必须设定  
 @Resource的name与生产接口定义中  
 @Output中的名称保持一致。