微服务架构实战学习笔记 第八章 Spring Cloud Stream与事件驱动

事件驱动架构定义了一个设计和实现运用系统的框架风格，事件可传输于松散耦合的服务和服务之间。事件驱动的架构基本组成包括事件发布（publish）、订阅（Subscribe）、消费等基本过程。

一、事件驱动架构与模型

1、基本时间驱动架构与实现机制

（1）引入事件驱动的目的是降低服务之间的耦合度、提升系统的扩展性。

在订单系统中，账户更新增加事件机制后的架构如图所示

（2）消息传递机制

消息传递机制能降低技术、空间和时间耦合，在消息传发送方和消息接收方增加了存储转发功能

2、事件驱动与领域模型

（1）事件订阅者模型

事件生命周期，一般存在生产、分发、存储和使用4个阶段。

1.简单订阅者：直接处理事件，表现为一个独立的事件处理程序，对应于事件的使用阶段。

2.即时转发订阅者:对应于事件的分发和使用阶段，具备简单订阅者功能，也可以把事件转发给其他订阅者。通常吧事件发送到消息队列。

3.事件存储订阅者:在处理事件的同时对事件进行持久化。引用于事件的存储和使用阶段。

2、领域事件

领域事件的基本结构通常包括唯一标识、生产时间、事件来源、时间类型以及各种业务数据，并且具有不可变行。

事件可以通过领域事件发布(DomainEventPublisher）者进行集中式存储，也可以分别保存在各个领域事件订阅者(DomainEventSubscriber)中

二、Spring Cloud Stream 

事件驱动架构的实现有两种解决方案：一种是直接使用RabbitMQ、kafka等消息中间件来实现消息传递，另一种是提供一个整体的平台解决方案，屏蔽各消息中间件技术实现的差异。

1、Spring Cloud Stream 基本结构

包含3中角色:消息发布者、消费者、消息传递系统本身

2、Spring  Cloud Stream 核心组件

具备4个核心组件，分别是Binder、Channel、Source、Sink，其中Binder和Channel 成对出现，而source和Sink分别面向消息发布者和消费者。

（1）Binder：是服务与消息传递系统之间的粘合剂，Stream实现了Kafka和RibbitMQ这两种中间建的Binder。可以动态的改变消息的目标地址、发送方式。

（2）Channel：Channel即通道，是队列（Queue）的一种抽象。队列的作用就是实现存储转发的媒介，消息发布者锁生成的消息都将保存在队列中并由消费者进行消费。通道的名称对应的就是队列的名称。

（3）Soure和Sink

Soure组件就是使用一个POJO(Plain Old  Java Object)对象来作为需要发布的消息，将该对象进行序列化后发布到通道中，而Sink组件用于监听通道并等待消息的到来，一旦有可用消息，Sink就会将该消息反序列化为一个POJO对象并用于处理业务逻辑。

3、Spring Cloud Stream 与 Spring Integration 

Spring Integration 是Spring 家族中的一员，作为轻量级、松耦合集成框架，与现有Spring 应用程序能够完美融合，是一种企业服务总线，能支持扩展主流系统集成模式，并提供众多基础性系统交互端点技术。

通常为两种基本表现形式，轮询PollableChannel和实现发布-订阅模式的SubscribaleChannel。都继承MessageChannel.

4、Spring Cloud Stream 与消息中间件

Spring Cloud Stream中间消息传递模型

（1）发布-订阅模式：Spring Cloud Stream中，统一通过发布-订阅模式完成消息的发布和消费。

（2）消费组：一旦使用了消费者组，则一条消息只能被同一个组中的某一个服务实例所消费。

（3）消息分区

保证同一消息始终由同一个消费者实例接受和处理。

1、Binder与消息中间件

Bider是中间层，负债与各种消息中间件交互。目前集成RabbitMQ和Kafka主流消息中间件。

（1）RabbitMQ：通过控制交换器（Exchange）与Queue之间的路由规则找到消息发送到那个Queue。

(2)Kafka

三、实现消息发布者

1、消息发送场景与实现流程

如账户信息变更

2、在服务中添加消息发布者

（1）初始化环境：引入spring-cloud-stream和spring-cloud-starter-strem-kafka 并在启动类中添加@EnableBinding(Source.class)注解。

（2）创建Event与Source

AccountChangeEvent事件包含:事件类型type、对应操作action、业务领域对象account

(3)配置Binder

需要在.yml配置文件中配置Binder信息。

(4)集成服务

四、实现消息消费者

1、消费场景与实现流程

AccountChangeEvent事件通过KafKa的accountChangeTopic发送消息到spring cloud Stream 中，Spring Cloud Strem 通过自定义的AccountChangeChannel获取消息并交由AccountchangeHandler实现消费逻辑。

2、在服务中添加消费者

（1）初始化消息消费环境，需要引入spring-cloud-stream 和 spring-cloud-starter-stream-KafKa这两个maven依赖。并在启动类上加入注解@EnableBinding(Sink.class)

(2)定义消息通道

（3）使用@StreamListener消费消息

3、分组消费

本文地址：https://blog.csdn.net/kyh1003381120/article/details/107309860