关于微服务:05期面向业务的消息服务落地实践

简介：传统的音讯队列对业务方提出了更高的要求，咱们冀望提供的是一种以业务为重心的，面向服务的解决方案。

这里记录的是学习分享内容，文章保护在 Github：studeyang/leanrning-share。

咱们在上次分享中聊到了畛域驱动设计和微服务，在 DDD 中有一个术语叫做畛域事件，例如订单模型中的订单已创立、商品已发货。畛域事件会触发下一步的业务操作，如果畛域事件产生在微服务内，能够通过观察者模式很容易实现音讯监听并解决。

<img src=”https://technotes.oss-cn-shenzhen.aliyuncs.com/2023/202303242122856.png” style=”zoom:50%;” />

如果产生在微服务之间，则需引入事件总线或者消息中间件。

一、音讯队列解决方案

通过技术选型后，咱们决定应用 Kafka 作为消息中间件，此时微服务间的通信示意图如下：

不过，间接应用音讯队列将面临以下问题：

1. 开发成本大：开发团队成员都须要对音讯队列如 Kafka 技术有肯定的理解，并且还须要关注连贯音讯队列的一些配置；
2. 治理难度大：各团队都应用一个音讯队列，其中一个团队应用不过后，例如创立了很多个 topic，造成资源节约；
3. 监控难度大：以后只有对 Kafka 集群简略的监控性能；
4. 运维艰难：遇到线上音讯没有生产时，很难排查问题，无从下手；
5. 降级难度大：Kafka-Client 须要降级时，波及到服务太多，导致降级老本高；

咱们冀望提供的是一种以业务为重心的，面向服务的解决方案。

也就是说，即便团队中没人理解音讯队列技术，也可能收发音讯。于是对 Kafka SDK 二次封装，次要就是为了简化音讯的接入，无需关注配置。

封装后解决了开发成本大、治理难度大的问题，然而离面向服务的解决方案指标还有肯定的差距。比方业务方监听到音讯后，执行一系列的业务逻辑异样了，想要做业务弥补，咱们的“基于 Kafka SDK 二次封装”的计划就没方法满足，只能要求音讯发送方再发一次音讯，但这又会影响其余音讯监听者。

于是咱们决定将音讯列队封装成音讯服务，对业务方提供切实的服务能力。

二、音讯服务解决方案

咱们熟知计算机中总线，在计算机系统中，不同的组件和设施须要互相通信以实现各种工作，此时，计算机总线就施展了重要作用。相似的，微服务零碎中，微服务就像是计算机系统中的各个组件和设施，而音讯服务充当的就是计算机总线的角色。音讯总线由此而来。

本文中呈现的音讯总线和音讯服务指的是同一个货色。

2.1 架构设计

发送音讯和接管音讯是音讯服务最根本的能力，这两项能力别离由音讯生产服务、音讯生产服务提供。

2.2 音讯的流转过程

三、音讯服务初体验

微服务架构采纳的技术栈是：SpringBoot、Kubernetes。

咱们将音讯总线取名为 Courier，Courier 的意思是“快递员”，音讯的传递相似于快递的收发，音讯总线正是快递员的角色。上面开始音讯服务的初体验。

3.1 零配置接入音讯总线

因为咱们的微服务应用的是 SpingBoot 来落地的，因而咱们提供了一个接入音讯总线的 spring-boot-starter。

<dependency>
    <groupId>com.casstime.open</groupId>
    <artifactId>courier-spring-boot-starter</artifactId>
</dependency>

接入音讯总线，微服务只须要一个 @EnableMessage 注解即可加载所有相干配置：

@EnableMessage
@SpringBootApplication
public class WebApplication {public static void main(String[] args) {SpringApplication.run(WebApplication.class, args);
    }
}

3.2 音讯构造定义

上面代码定义了一个音讯的根本信息，也称为音讯 Header，包含音讯 id，分区键 primaryKey，起源服务 service，音讯 topic，创立工夫 timstamp。

public abstract class Message {
    private String id;
    private String primaryKey;
    private String service;
    private String topic;
    private Date timeStamp;
}

音讯能够分为两类，一类是事件，另一类是播送。定义如下：

// 事件
public abstract class Event extends Message {}

// 播送
public abstract class Event extends Message {}

业务音讯内容称为音讯 Body，例如订单已创立这个音讯体的定义：

@Topic(name = "order")
public class OrderCreated extends Event {
    private String orderId;
    private String orderName;
    private Date createdAt;
}

3.3 使音讯收发变得简略

业务方能够在业务执行办法的任一处，只须要一行代码，即可实现音讯的发送。

// 发送音讯
EventPublisher.publish(new OrderCreated());

对于音讯的监听，业务方只需关注业务逻辑的执行，屏蔽了 Offset 提交、重试等技术实现。

// 接管音讯
@EventHandler(topic = "order", consumerGroup = "consumer-group1")
public class OrderMessageHandler {public void handle(OrderCreated orderCreated) {System.out.println("receive message:" + orderCreated);
    }
}

3.4 提供 5 种性能类型的音讯

咱们提供了 5 种不同性能类型的音讯，满足各类业务场景。

1、事件音讯

@Topic(name = "order")
public class OrderCreated extends Event {
    private String orderId;
    private String orderName;
    private Date createdAt;
}

public void send() {EventPublisher.publish(new OrderCreated());
}

下面音讯定义是事件，这是应用最多的一种音讯。

2、播送音讯

播送音讯的生产示意图如下：

@Topic(name = "order")
public class CacheUpdate extends Broadcast {
    private String orderId;
    private String orderName;
    private Date createdAt;
}

public void send() {EventPublisher.publish(new CacheUpdate());
}

下面音讯定义时，继承了Broadcast，示意这是一个播送音讯，生产服务的每个节点都将会收到这个播送。例如更新本地缓存事件，就须要用到播送音讯。

3、程序音讯

@Topic(name = "order")
public class OrderCreated extends Event {
    @PrimaryKey
    private String orderId;
    private String orderName;
    private Date createdAt;
}

public void send() {EventPublisher.publish(new OrderCreated());
}

下面音讯定义时，在 orderId 上加了 @PrimaryKey 注解，示意雷同 orderId 的音讯会有序的生产。

4、事务音讯

@Topic(name = "order")
public class OrderCreated extends Event {
    private String orderId;
    private String orderName;
    private Date createdAt;
}

@Transactional
public void send() {EventPublisher.publish(new OrderCreated());
}

下面音讯发送时，在办法上增加了 @Transactional 注解，这是 Spring 的注解，示意这个办法里的逻辑执行是有事务性的。

5、提早音讯

@Topic(name = "order")
public class OrderCreated extends Event {
    private String orderId;
    private String orderName;
    private Date createdAt;
}

@Transactional
public void send() {EventPublisher.publish(new OrderCreated(), 2, TimeUnit.SECONDS);
}

下面音讯发送多了两个参数，示意提早 2 秒接管。

3.5 音讯追踪

只有是通过 EventPublisher.publish() 办法发送的音讯，都能够追踪到这条音讯记录。

音讯定义了 5 种状态：

• 发送失败（SEND_FAIL）：通常音讯定义不标准，音讯体过大；多数因为网络抖动。
• 已提交（COMMITED）：音讯总线已收到音讯。
• 推送失败（PUSH_FAIL）：例如服务已下线。
• 解决失败（HANDLE_FAIL）：监听到了音讯，然而执行业务逻辑抛出了异样。
• 已解决（HANDLED）

作为音讯的发送方，关注的是音讯是否发送胜利，可通过上面页面查问。

作为音讯的接管方，关注的是音讯是否失常生产，可通过上面页面查问。

3.6 音讯高牢靠

对于 5 种状态的音讯，解决策略如下：

• 发送失败（SEND_FAIL）：主动重试 + 手动重试，可在音讯管理中心手动再发送。
• 已提交（COMMITED）：长期解决已提交状态的音讯，可能生产方已接管，但状态流转异样，音讯总线会定时重试。
• 推送失败（PUSH_FAIL）：主动重试 + 提早重试。
• 解决失败（HANDLE_FAIL）：主动重试默认敞开，由生产方决定是否开启重试。
• 已解决（HANDLED）：也可手动重试。

封面

相干文章

兴许你对上面文章也感兴趣。

• 04 期：畛域驱动设计与微服务
• 学习分享（第 3 期）：你所了解的架构是什么？