Spring Cloud 之 Stream.

简介: Spring Cloud 之 Stream.一、简介Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。Spring Cloud Stream 为一些供应商的消息中间件产品(目前集成了 RabbitMQ 和 Kafka)提供了个性化的自动化配置实现,并且引入了发布/订阅、消费组以及消息分区这三个核心概念。

Spring Cloud 之 Stream.
一、简介
Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。

Spring Cloud Stream 为一些供应商的消息中间件产品(目前集成了 RabbitMQ 和 Kafka)提供了个性化的自动化配置实现,并且引入了发布/订阅、消费组以及消息分区这三个核心概念。简单地说,Spring Cloud Stream 本质上就是整合了 Spring Boot 和 Spring Integration, 实现了一套轻量级的消息驱动的微服务框架。

通过使用 Spring Cloud Stream,可以忽略消息中间件的差异,有效简化开发人员对消息中间件的使用复杂度,让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。

回到顶部
二、快速入门

  1. pom.yml

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
    </dependencies>
  2. application.yml
    配置消息中间件的连接信息:

spring:
application:

name: cloud-stream

rabbitmq:

host: 127.0.0.1
port: 5672
username: guest
password: guest
  1. 消息监听/消费
    @EnableBinding({Source.class, Sink.class})

public class SinkReceiver {

private Logger log = LoggerFactory.getLogger(SinkReceiver.class);

@StreamListener(Sink.INPUT)
@SendTo(Source.OUTPUT)
public Object processInput(String message) {
    log.info("Input Stream Receiver:{}", message);
    return message;
}

@StreamListener(Source.OUTPUT)
public void processOutPut(String message) {
    log.info("Output Stream Receiver:{}", message);
}

}
@EnableBinding:实现对消息通道(Channel) 的绑定,其中 Sink 是 Spring Cloud Stream 默认的输入通道,Source 是 Spring Cloud Stream 中默认的输出通道。
@StreamListener:将被修饰的方法注册为消息中间件上数据流的事件监听器,注解中的属性值对应了监听的消息通道名。如果不设置属性值,将默认使用方法名作为消息通道名。
@SendTo:很多时候在处理完输入消息之后, 需要反馈一个消息给对方, 这时候可以通过 @SendTo 注解来指定返回内容的输出通道。
4.消息生产
消息生产有两种方式,一种是利用注入消息通道来发送消息,如下:

@RunWith(SpringRunner.class)
@SpringBootTest(classes = StreamApplication.class)
public class SinkOutputTest {

@Autowired
private Sink sink;
@Autowired
private Source source;

@Test
public void sink() {
    sink.input().send(MessageBuilder.withPayload("From SinkSender").build());
}

@Test
public void source() {
    source.output().send(MessageBuilder.withPayload("From SourceSender").build());
}

}
另外一种是使用 Spring Integration 的原生支持 — @InboundChannelAdapter

@EnableBinding(value = {Source.class})
@SpringBootApplication
public class StreamApplication {

public static void main(String[] args) {
    SpringApplication.run(StreamApplication.class, args);
}

@Bean
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000"))
public MessageSource<String> timerMessageSource() {
    return () -> new GenericMessage<>("2019/08/06");
}

}
回到顶部
三、绑定器
Spring Cloud Stream 构建的应用程序与消息中间件之间是通过绑定器 Binder 相关联的,绑定器对于应用程序而言起到了隔离作用, 它使得不同消息中间件的实现细节对应用程序来说是透明的。所以对于每一个 Spring Cloud Stream 的应用程序来说, 它不需要知晓消息中间件的通信细节,它只需知道 Binder 对应程序提供的抽象概念来使用消息中间件来实现业务逻辑即可,而这个抽象概念就是在快速入门中我们提到的消息通道:Channel。如下图所示,在应用程序和 Binder 之间定义了两条输入通道和三条输出通道来传递消息,而绑定器则是作为这些通道和消息中间件之间的桥梁进行通信。

通过定义绑定器作为中间层,完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的 Channel 通道,使得应用程序不需要再考虑各种不同的消息中间件的实现。当需要升级消息中间件,或是更换其他消息中间件产品时,我们要做的就是更换它们对应的 Binder 绑定器而不需要修改任何 SpringBoot 的应用逻辑。

回到顶部
四、消费组
Spring Cloud Stream中的消息通信方式遵循了发布-订阅模式,当一条消息被投递到消息中间件之后,它会通过共享的 Topic 主题进行广播,消息消费者在订阅的主题中收到它并触发自身的业务逻辑处理。(这里提到的 Topic 指的是 Stream 的抽象概念,可以是 RabbitMQ 中的 Exchange,也可以是 Kafka 中的 Topic)。

发布-订阅模式会带来一个问题。因为在微服务架构中,我们的每一个微服务应用为了实现高可用和负载均衡, 实际上都会部署多个实例。按照消息广播的性质,多个实例都会接收到消息,从而导致重复消费。为了解决这个问题, 在Spring Cloud Stream中提供了消费组的概念。

如果在同一个主题上的应用需要启动多个实例的时候,我们可以通过 spring.cloud.stream.bindings..group 属性为应用指定一个组名,这样这个应用的多个实例在接收到消息的时候,只会有一个成员真正收到消息并进行处理。

spring:
application:

name: cloud-stream

cloud:

stream:
  bindings:
    input:
      # 设置消费组,保证只有一个实例消费到消息
      # 如果不设置消费组,Stream 将会为每个实例生成一个消费组
      group: ${spring.application.name}

回到顶部
五、消息分区
通过引入消费组的概念,我们已经能够在多实例的清况下,保障每个消息只被组内的一个实例消费。但是消费组无法控制消息具体被哪个实例消费。也就是说,对于同一条消息,它多次到达之后可能是由不同的实例进行消费的。但是对于一些业务场景,需要对一些具有相同特征的消息设置每次都被同一个消费实例处理。

消息分区的引入就是为了解决这样的问题:当生产者将消息数据发送给多个消费者实例时,保证拥有共同特征的消息数据始终是由同一个消费者实例接收和处理。

消费者分区:

spring:
application:

name: cloud-stream

cloud:

stream:
  instance-count: 1
  instance-index: 0
  bindings:
    input:
      consumer:
        partitioned: true

spring.cloud.stream.bindings.input.consumer.partitioned = true 开启消费者分区功能。
spring.cloud.stream.instance-count = 1 当前消费者的总实例个数,即应用程序部署的实例数量。
spring.cloud.stream.instance-index = 0 当前实例的索引号,从 0 开始,最大为 -1 。用于消息生产的时候锁定该实例。(消息生产的时候 "hashCode(key) % partitionCount" 的计算值等于该设置的值,即转发到该实例上)
生产者分区:

spring:
application:

name: cloud-stream

cloud:

stream:
  bindings:
    output:
      producer:
        partitionCount: 1
        partitionKeyExtractorName: keyStrategy

spring.cloud.stream.bindings.output.producer.partitionCount = 1 ,消息生产需要广播的消费者数量。即消息分区的数量。
spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName = keyStrategy ,Spring Bean — 用来消息的特征值计算。(分区选择计算规则为 "hashCode(key) % partitionCount" , 这里的 key 根据 partitionKeyExpression 或 partitionKeyExtractorName 的配置计算得到)
@Component
public class KeyStrategy implements PartitionKeyExtractorStrategy {


@Override
public Object extractKey(Message<?> message) {
    return message.getPayload();
}

}

演示源代码:https://github.com/JMCuixy/spring-cloud-demo
内容参考:《Spring Cloud 微服务实战》
原文地址https://www.cnblogs.com/jmcui/p/11279388.html

相关文章
|
5月前
|
监控 负载均衡 Java
深入理解Spring Cloud中的服务网关
深入理解Spring Cloud中的服务网关
|
23天前
|
消息中间件 监控 Java
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
31 6
|
23天前
|
Java 关系型数据库 MySQL
如何将Spring Boot + MySQL应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot + MySQL应用程序部署到Pivotal Cloud Foundry (PCF)
41 5
|
23天前
|
缓存 监控 Java
如何将Spring Boot应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot应用程序部署到Pivotal Cloud Foundry (PCF)
31 5
|
5月前
|
设计模式 监控 Java
解析Spring Cloud中的断路器模式原理
解析Spring Cloud中的断路器模式原理
|
5月前
|
资源调度 Java 调度
Spring Cloud Alibaba 集成分布式定时任务调度功能
Spring Cloud Alibaba 发布了 Scheduling 任务调度模块 [#3732]提供了一套开源、轻量级、高可用的定时任务解决方案,帮助您快速开发微服务体系下的分布式定时任务。
15055 32
|
5月前
|
负载均衡 Java Spring
Spring cloud gateway 如何在路由时进行负载均衡
Spring cloud gateway 如何在路由时进行负载均衡
589 15
|
5月前
|
Java Spring
spring cloud gateway在使用 zookeeper 注册中心时,配置https 进行服务转发
spring cloud gateway在使用 zookeeper 注册中心时,配置https 进行服务转发
127 3
|
4月前
|
消息中间件 Java 开发工具
【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常
【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常