2022年7月7日

RocketMQ 消息集成:多类型业务消息-普通消息
引言 CDC(Change Data Capture)指的是监听上游数据变更,并将变更信息同步到下游业务以供进一步处理的一种应用场景。近年来事件驱动架构(EDA)热度逐步上升,日渐成为项目架构设计者的第一选择。EDA 天然契合 CDC 的底层基础架构,其将数据变更作为事件,各个服务通过监听自己感兴趣的事件来完成一些列业务驱动。阿里云 EventBridge 是阿里云推出的一款无服务器事件总线服务,能够帮助用户轻松快捷地搭建基于 EDA 架构的应用。近期,EventBridge 事件流已经支持了基于阿里云 DTS[1]服务的 CDC 能力。本文将从 CDC、CDC 在 EventBridge 上的应用以及若干最佳实践场景等方面,为大家介绍如何利用 EventBridge 轻松构建 CDC 应用。 CDC 概述 基本原理与应用场景 CDC 从源数据库捕获增量的数据以及数据模式变更,以高可靠、低延时的数据传输将这些变更有序地同步到目标数据库、数据湖或者其他数据分析服务。目前业界主流的开源 CDC 工具包括 Debezium[2]、Canal[3] 以及 Maxwell[4]。 图片来源: 目前业界主要有以下几类 CDC 的实现: 1. 基于时间戳或版本号 基于时间戳的方式要求数据库表有一个字段代表更新时间戳,当存在数据插入或更新时,对应时间戳字段就会随之更新。CDC 组件周期性检索更新时间大于上次同步时间的数据记录,即可捕获本周期内数据的变更。基于版本号跟踪和基于时间戳跟踪原理基本一致,要求开发者变更数据时必须更新数据的版本号。 2. 基于快照 基于快照的 CDC 实现在存储层面使用到了数据源 3 份副本,分别是原始数据、先前快照和当前快照。通过对比 2 次快照之间的差异来获取这之间的数据变更内容。 3. 基于触发器 基于触发器的 CDC 实现方式事实上是在源表上建立触发器将对数据的变更操作(INSERT、UPDATE、DELETE)记录存储下来。例如专门建立一张表记录用户的变更操作,随后创建 INSERT、UPDATE、DELETE 三种类型的触发器将用户变更同步到此表。 4. 基于日志 以上三种方式都对源数据库存在一定侵入性,而基于日志的方式则是一种非侵入性的 CDC 方式。数据库利用事务日志实现灾备,例如 MySQL 的 binlog 就记录了用户对数据库的所有变更操作。基于日志的 CDC 通过持续监听事务日志来实时获取数据库的变化情况。 CDC 的应用场景广泛,包括但不限于这些方面:异地机房数据库同步、异构数据库数据同步、微服务解耦、缓存更新与 CQRS 等。 基于阿里云的 CDC 解决方案:DTS 数据传输服务 DTS(Data Transmission Service)是阿里云提供的实时数据流服务,支持关系型数据库(RDBMS)、非关系型的数据库(NoSQL)、数据多维分析(OLAP)等数据源间的数据交互,集数据同步、迁移、订阅、集成、加工于一体。其中,DTS 数据订阅[5]功能可以帮助用户获取自建 MySQL、RDS MySQL、Oracle 等数据库的实时增量数据。 CDC 在EventBrige上的应用 阿里云 EventBridge 提供了事件总线[6]与事件流[7] 2 款不同应用场景的事件路由服务。 事件总线底层拥有事件的持久化能力,可以按照需要将事件路由到多个事件目标中。 事件流适用于端到端的流式数据处理场景,对源端产生的事件实时抽取、转换和分析并加载至目标端,无需创建事件总线,端到端转储效率更高,使用更轻便。 为了更好地支持用户在 CDC 场景下的需求,EventBridge 在事件流源端支持了阿里云 DTS 的数据订阅功能,用户仅需简单配置,即可将数据库变更信息同步到 EventBridge 事件流。 EventBridge 定制了基于 DTS sdk 的 DTS Source Connector。当用户配置事件提供方为 DTS 的事件流时,source connector 会实时地从 DTS 服务端拉取 DTS record 数据。数据拉取到本地后,会进行一定的结构封装,保留 id、operationType、topicPartition、beforeImage、afterImage 等数据,同时增加 streaming event 所需要的一些系统属性。 DTS Event 样例可参考 EventBridge 官方文档 EventBridge Streaming 保证了 DTS 事件的顺序性,但存在事件重复投递的可能性,EventId 在保证了和每条 DTS record 的一一映射关系,用户可依据此字段来对事件做幂等处理。 创建源为 DTS 的 EventBridge 事件流 下面展示如何在 EventBridge 控制台创建源为 DTS 的事件流 前期准备   1. 开通 EventBridge 服务; 2. 创建 DTS 数据订阅任务; 3. 创建用于消费订阅数据的消费组账号信息。 创建事件流   1. 登陆 EventBridge 控制台,点击左侧导航栏,选择“事件流”,在事件流列表页点击“创建事件流”; 2. “基本信息”中“事件流名称”与“描述”按照需要填写即可; 3. 在创建事件流,选择事件提供方时,下拉框选择“数据库 DTS”; 4. 在“数据订阅任务”一栏中选择已创建的 DTS 数据订阅任务。在消费组一栏,选择要使用哪个消费组消费订阅数据,同时填写消费组密码与初始消费时间。 5. 事件流规则与目标按照需要填写,保存启动即可创建以 DTS 数据订阅为事件源的事件流。 注意事项 使用时有以下几点需要注意: 1. EventBridge 使用的是 SUBSCRIBE 消费模式[8],所以请保证当前 DTS 消费组没有其他客户端实例在运行。如果设置的消费组在之前有运行,则传入的位点失效,会基于此消费组上次消费过的位点继续消费; 2. 创建 DTS 事件源时传入的位点仅在新消费组第一次运行时起效,后续任务重启后会基于上次消费位点继续消费; 3. EventBridge 事件流订阅 OperationType 为 INSERT、DELETE、UPDATE、DDL 类型的 DTS 数据; 4. 使用 DTS  事件源可能会有消息重复,即保证消息不丢,但无法保证仅投递一次,建议用户做好幂等处理; 5.用户如果需要保证顺序消费,则需要将异常容忍策略设置为“NONE”,即不容忍异常。在这种情况下,如果事件流目标端消费消息异常,整个事件流将暂停,直至恢复目标端正常。 最佳实践示例 基于EventBridge 实现 CQRS 在 CQRS(Command Query Responsibility Segregation)模型中,命令模型用于执行写以及更新操作,查询模型用于支持高效的读操作。读操作和写操作使用的数据模型存在一定区别,需要使用一定方式保证数据的同步,基于 EventBridge 事件流的 CDC 可以满足这样的需求。 基于云上服务,用户可以使用如下方式轻松构建基于 EventBridge 的 CQRS: 1. 命令模型操作数据库进行变更,查询模型读取 elasticsearch 获取数据; 2. 开启 DTS 数据订阅任务,捕获 DB 变更内容; 3.配置 EventBridge 事件流,事件提供方为 DTS 数据订阅任务,事件接收方为函数计算 FC; 4. FC 中的服务即为更新 elasticsearch 数据操作。 微服务解耦 CDC 也可以用于微服务解耦。例如下文是一个电商平台的订单处理系统,当有新建的未付款订单产生时,数据库会有一条 INSERT 操作,而当某笔订单状态由“未付款”变为“已付款”时,数据库会有一条 UPDATE 操作。根据订单状态变化的不同,后端会有不同的微服务来对此进行处理。 1. 用户下单/付款,订单系统进行业务处理,将数据变更写入 DB; 2. 新建 DTS 订阅任务捕获 DB 数据变更; 3. 搭建 EventBridge 事件流。事件提供方为 DTS 数据订阅任务,事件接收方为 RocketMQ; 4. 在消费 RocketMQ 数据时,同一个 topic 下启用 3 个 group 代表不同的业务消费逻辑; a. GroupA 将捕获到的 DB 变更用户缓存更新,便于用户查询订单状态; b. GroupB 下游关联财务系统,仅处理新建订单,即处理 DB 操作类型为 INSERT 的事件,丢弃其余类型事件; c. GroupC 仅关心订单状态由“未付款”变为“已付款”的事件,当有符合条件事件到达时,调用下游物流、仓储系统,对订单进行进一步处理。 如果采用接口调用方式,那么用户在下单之后订单系统将分别需要调用缓存更新接口、新建订单接口以及订单付款接口,业务耦合性过高。除此之外,这种模式使得数据消费端不用担心上游订单处理接口返回内容的语义信息,在存储模型不变的情况下,直接从数据层面判断此次数据变更是否需要处理以及需要怎样的处理。同时,消息队列天然的消息堆积能力也可以帮助用户在订单峰值到来时实现业务削峰填谷。 事实上,目前 EventBridge Streaming 支持的消息产品还包括 RabbitMQ、Kafka、MNS 等,在实际操作中用户可以根据自己的需要进行选择。 数据库备份&异构数据库同步 数据库灾备和异构数据库数据同步也是 CDC 重要的应用场景。使用阿里云 EventBridge 亦可以快速搭建此类应用。 1. 新建 DTS 数据订阅任务,捕获用户 MySQL 数据库变更; 2. 搭建 EventBridge 事件流,事件提供方为 DTS 数据订阅任务; 3. 使用 EventBridge 在目的数据库执行指定 sql,实现数据库备份; 4. 数据变更事件投递到函数计算,用户业务根据数据变化内容更新对应异构数据库。 自建 SQL 审计 对于用户有自建 SQL 审计的需求,使用 EventBridge 也可以轻松实现。 1. 新建 DTS 数据订阅任务,捕获数据库变更; 2. 搭建 EventBridge 事件流,事件提供方为 DTS,事件接收方为日志服务 SLS; 3. 用户需要对 SQL 进行审计时,通过查询 SLS 进行。 总结 本文介绍了 CDC 的一些概念、CDC 在 EventBridge 上的应用以及若干最佳实践场景。随着支持产品的不断增加,EventBridge 所承载的生态版图也不断扩大,从消息生态到数据库生态,从日志生态到大数据生态,EventBridge 不断扩大其适用领域,巩固云上事件枢纽的地位,此后也将按照这个方向继续发展,技术做深,生态做广。 _参考链接:_ _[1] DTS:_ _[2] Debezium:_ _[3] Canal:_ _[4] Maxwell:_ _[5] DTS 数据订阅:_ _[6] 事件总线:_ _[7] 事件流:_ _[8] SUBSCRIBE 消费模式:_ 活动推荐 阿里云基于 Apache RocketMQ 构建的企业级产品消息队列RocketMQ 5.0版现开启活动: 1、新用户首次购买包年包月,即可享受全系列 85折优惠! 了解活动详情:
作者:昶风
#技术探索 #功能特性

2022年1月14日

全链路灰度之 RocketMQ 灰度
之前的系列文章中,我们已经通过全链路金丝雀发布这个功能来介绍了 MSE 对于全链路流量控制的场景,我们已经了解了 Spring Cloud 和 Dubbo 这一类 RPC 调用的全链路灰度应该如何实现,但是没有涉及到消息这类异步场景下的流量控制,今天我们将以上次介绍过的《》中的场景为基础,来进一步介绍消息场景的全链路灰度。 虽然绝大多数业务场景下对于消息的灰度的要求并不像 RPC 的要求得这么严格,但是在以下两个场景下,还是会对消息的全链路有一定的诉求的。 1、第一种场景是在消息消费时,可能会产生新的 RPC 调用,如果没有在消息这一环去遵循之前设定好的全链路流量控制的规则,会导致通过消息产生的这部分流量“逃逸”,从而导致全链路灰度的规则遭到破坏,导致出现不符合预期的情况。 为了防止出现这个情况,我们需要在消费时候将消息里原来的流量标复原,并在 RPC 调用的时候遵循原来的规则。我们通过架构图来详细描述一下,满足这个逻辑之后,调用链路是怎样的,从下图中我们可以看到,灰度和基线环境生产出来的消息,虽然在消息推送的时候是随机的,但是在消费过程中,产生的新的 RPC 调用,还是能够回到流量原来所属的环境。 2、第二种场景需要更加严格的消息灰度隔离。比如当消息的消费逻辑进行了修改时,这时候希望通过小流量的方式来验证新的消息消费逻辑的正确性,要严格地要求灰度的消息只能被推送给灰度的消息消费者。 今天我们就来实操一下第二种场景消息的全链路灰度,目前 MSE 仅支持 RocketMQ 消息的灰度。若您使用的是开源版 RocketMQ,那么版本需要在 4.5.0 及以上,若您使用的是阿里云商业版 RocketMQ,那么需要使用铂金版,且 Ons Client 版本在 1.8.0.Final 及以上。如果只是想使用第一种场景,只需要给 B 应用开启全链路灰度的功能即可,不需要做额外的消息灰度相关的配置。 在这次最佳实践的操作中,我们是将应用部署在阿里云容器服务 Kubernetes 版本,即 ACK 集群来演示,但是事实上,消息灰度对于应用的部署模式是没有限制性要求的,您可以参考 MSE 帮助文档,找到自己所使用的部署模式对应的接入方式,也能使用消息全链路灰度。 前提条件 1. 开通 MSE 专业版,请参见开通 MSE 微服务治理专业版[1]。 2. 创建 ACK 集群,请参见创建 Kubernetes 集群[2]。 操作步骤 步骤一:接入 MSE 微服务治理 1、安装 mseackpilot 1. 登录容器服务控制台[3]。 2. 在左侧导航栏单击市场 应用目录。 3. 在应用目录页面点击阿里云应用,选择微服务,并单击 ackmsepilot。 4. 在 ackmsepilot 页面右侧集群列表中选择集群,然后单击创建。 安装 MSE 微服务治理组件大约需要 2 分钟,请耐心等待。 创建成功后,会自动跳转到目标集群的 Helm 页面,检查安装结果。如果出现以下页面,展示相关资源,则说明安装成功。 2、为 ACK 命名空间中的应用开启 MSE 微服务治理 1. 登录 MSE 治理中心控制台[4],如果您尚未开通 MSE 微服务治理,请根据提示开通。 2. 在左侧导航栏选择微服务治理中心 Kubernetes 集群列表。 3. 在 Kubernetes 集群列表页面搜索框列表中选择集群名称或集群 ID,然后输入相应的关键字,单击搜索图标。 4. 单击目标集群操作列的管理。 5. 在集群详情页面命名空间列表区域,单击目标命名空间操作列下的开启微服务治理。 6. 在开启微服务治理对话框中单击确认。 步骤二:还原线上场景 首先,我们将分别部署  springcloudzuul、springclouda、springcloudb、springcloudc 这四个业务应用,以及注册中心 Nacos Server 和消息服务 RocketMQ Server,模拟出一个真实的调用链路。 Demo 应用的结构图下图,应用之间的调用,既包含了 Spring Cloud 的调用,也包含了 Dubbo 的调用,覆盖了当前市面上最常用的两种微服务框架。其中 C 应用会生产出 RocketMQ 消息,由 A 应用进行消费,A 在消费消息时,也会发起新的调用。这些应用都是最简单的 Spring Cloud 、 Dubbo 和 RocketMQ 的标准用法,您也可以直接在 项目上查看源码。 部署之前,简单介绍一下这个调用链路 springcloudzuul 应用在收到 “/A/dubbo” 的请求时,会把请求转发给 springclouda ,然后 springclouda 通过 dubbo 协议去访问 springcloudb, springcloudb 也通过 dubbo 协议去访问 springcloudc,springcloudc 在收到请求后,会生产一个消息,并返回自己的环境标签和 ip。这些生产出来的消息会由 springclouda 应用消费,springclouda 应用在消费消息的时候,会通过 spring cloud 去调用 B,B 进而通过 spring cloud 去调用 C,并且将结果输出到自己的日志中。 当我们调用 /A/dubbo 的时候 返回值是这样 A[10.25.0.32] B[10.25.0.152] C[10.25.0.30] 同时,A 应用在接收到消息之后,输出的日志如下 20211228 10:58:50.301 INFO 1 [essageThread_15] c.a.mse.demo.service.MqConsumer : topic:TEST_MQ,producer:C[10.25.0.30],invoke result:A[10.25.0.32] B[10.25.0.152] C[10.25.0.30] 熟悉了调用链路之后,我们继续部署应用,您可以使用 kubectl 或者直接使用 ACK 控制台来部署应用。部署所使用的 yaml 文件如下,您同样可以直接在 上获取对应的源码。 部署 Nacos Server apiVersion: apps/v1 kind: Deployment metadata: name: nacosserver spec: selector: matchLabels: app: nacosserver template: metadata: annotations: labels: app: nacosserver spec: containers: env: name: MODE value: "standalone" image: registry.cnshanghai.aliyuncs.com/yizhan/nacosserver:latest imagePullPolicy: IfNotPresent name: nacosserver ports: containerPort: 8848 apiVersion: v1 kind: Service metadata: name: nacosserver spec: type: ClusterIP selector: app: nacosserver ports: name: http port: 8848 targetPort: 8848 部署业务应用 apiVersion: apps/v1 kind: Deployment metadata: name: springcloudzuul spec: selector: matchLabels: app: springcloudzuul template: metadata: annotations: msePilotCreateAppName: springcloudzuul labels: app: springcloudzuul spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre name: enable.mq.invoke value: 'true' image: registry.cnshanghai.aliyuncs.com/yizhan/springcloudzuul:1.0.0 imagePullPolicy: Always name: springcloudzuul ports: containerPort: 20000 apiVersion: v1 kind: Service metadata: annotations: service.beta.kubernetes.io/alibabacloudloadbalancerspec: slb.s1.small service.beta.kubernetes.io/alicloudloadbalanceraddresstype: internet name: zuulslb spec: ports: port: 80 protocol: TCP targetPort: 20000 selector: app: springcloudzuul type: LoadBalancer status: loadBalancer: {} apiVersion: apps/v1 kind: Deployment metadata: name: springclouda spec: selector: matchLabels: app: springclouda template: metadata: annotations: msePilotCreateAppName: springclouda labels: app: springclouda spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre image: registry.cnshanghai.aliyuncs.com/yizhan/springclouda:1.0.0 imagePullPolicy: Always name: springclouda ports: containerPort: 20001 livenessProbe: tcpSocket: port: 20001 initialDelaySeconds: 10 periodSeconds: 30 apiVersion: apps/v1 kind: Deployment metadata: name: springcloudb spec: selector: matchLabels: app: springcloudb template: metadata: annotations: msePilotCreateAppName: springcloudb labels: app: springcloudb spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre image: registry.cnshanghai.aliyuncs.com/yizhan/springcloudb:1.0.0 imagePullPolicy: Always name: springcloudb ports: containerPort: 20002 livenessProbe: tcpSocket: port: 20002 initialDelaySeconds: 10 periodSeconds: 30 apiVersion: apps/v1 kind: Deployment metadata: name: springcloudc spec: selector: matchLabels: app: springcloudc template: metadata: annotations: msePilotCreateAppName: springcloudc labels: app: springcloudc spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre image: registry.cnshanghai.aliyuncs.com/yizhan/springcloudc:1.0.0 imagePullPolicy: Always name: springcloudc ports: containerPort: 20003 livenessProbe: tcpSocket: port: 20003 initialDelaySeconds: 10 periodSeconds: 30 apiVersion: apps/v1 kind: Deployment metadata: name: rockectmqbroker spec: selector: matchLabels: app: rockectmqbroker template: metadata: labels: app: rockectmqbroker spec: containers: command: sh mqbroker 'n' 'mqnamesrv:9876' 'c /home/rocketmq/rocketmq4.5.0/conf/broker.conf' env: name: ROCKETMQ_HOME value: /home/rocketmq/rocketmq4.5.0 image: registry.cnshanghai.aliyuncs.com/yizhan/rocketmq:4.5.0 imagePullPolicy: Always name: rockectmqbroker ports: containerPort: 9876 protocol: TCP containerPort: 10911 protocol: TCP containerPort: 10912 protocol: TCP containerPort: 10909 apiVersion: apps/v1 kind: Deployment metadata: name: rocketmqnameserver spec: selector: matchLabels: app: rocketmqnameserver template: metadata: labels: app: rocketmqnameserver spec: containers: command: sh mqnamesrv env: name: ROCKETMQ_HOME value: /home/rocketmq/rocketmq4.5.0 image: registry.cnshanghai.aliyuncs.com/yizhan/rocketmq:4.5.0 imagePullPolicy: Always name: rocketmqnameserver ports: containerPort: 9876 protocol: TCP containerPort: 10911 protocol: TCP containerPort: 10912 protocol: TCP containerPort: 10909 protocol: TCP apiVersion: v1 kind: Service metadata: name: mqnamesrv spec: type: ClusterIP selector: app: rocketmqnameserver ports: name: mqnamesrv98769876 port: 9876 targetPort: 9876 安装成功后,示例如下: ➜ ~ kubectl get svc,deploy NAME TYPE CLUSTERIP EXTERNALIP PORT(S) AGE service/kubernetes ClusterIP 192.168.0.1 443/TCP 7d service/mqnamesrv ClusterIP 192.168.213.38 9876/TCP 47h service/nacosserver ClusterIP 192.168.24.189 8848/TCP 47h service/zuulslb LoadBalancer 192.168.189.111 123.56.253.4 80:30260/TCP 47h NAME READY UPTODATE AVAILABLE AGE deployment.apps/nacosserver 1/1 1 1 4m deployment.apps/rockectmqbroker 1/1 1 1 4m deployment.apps/rocketmqnameserver 1/1 1 1 5m deployment.apps/springclouda 1/1 1 1 5m deployment.apps/springcloudb 1/1 1 1 5m deployment.apps/springcloudc 1/1 1 1 5m deployment.apps/springcloudzuul 1/1 1 1 5m 同时这里我们可以通过 zuulslb 来验证一下刚才所说的调用链路 ➜ ~ curl http://123.56.253.4/A/dubbo A[10.25.0.32] B[10.25.0.152] C[10.25.0.30] 步骤三:开启消息灰度功能 现在根据控制台的提示,在消息的生产者 springcloudc 和消息的消费者 springclouda 都开启消息的灰度。我们直接通过 MSE 的控制台开启,点击进入应用的详情页,选择“消息灰度”标签。 可以看到,在未打标环境忽略的标签中,我们输入了 gray,这里意味着,带着 gray 环境标的消息,只能由 springcloudagray 消费,不能由 springclouda 来消费。 _1、这里需要额外说明一下,因为考虑到实际场景中,springcloudc 应用和 springclouda  应用的所有者可能不是同一个人,不一定能够做到两者同时进行灰度发布同步的操作,所以在消息的灰度中,未打标环境默认的行为是消费所有消息。这样 springcloudc 在进行灰度发布的时候,可以不需要强制 springclouda 应用也一定要同时灰度发布。_ _2、我们把未打标环境消费行为的选择权交给 springclouda 的所有者,如果需要实现未打标环境不消费 cgray 生产出来的消息,只需要在控制台进行配置即可,配置之后实时生效。_ 使用此功能您无需修改应用的代码和配置。 消息的生产者和消息的消费者,需要同时开启消息灰度,消息的灰度功能才能生效。 消息类型目前只支持 RocketMQ,包含开源版本和阿里云商业版。 如果您使用开源 RocketMQ,则 RocketMQ Server 和 RocketMQ Client 都需要使用 4.5.0 及以上版本。 如果您使用阿里云 RocketMQ,需要使用铂金版,且 Ons Client 使用 1.8.0.Final 及以上版本。 开启消息灰度后,MSE 会修改消息的 Consumer Group。例如原来的 Consumer Group 为 group1,环境标签为 gray,开启消息灰度后,则 group 会被修改成 group1_gray,如果您使用的是阿里云 RocketMQ ,请提前创建好 group。 默认使用 SQL92 的过滤方式,如果您使用的开源 RocketMQ,需要在服务端开启此功能(即在 broker.conf 中配置 enablePropertyFilter=true)。 默认情况下,未打标节点将消费所有环境的消息,若需要指定 未打标环节点 不消费 某个标签环境生产出来的消息,请配置“未打标环境忽略的标签”,修改此配置后动态生效,无需重启应用。 步骤四:重启节点,部署新版本应用,并引入流量进行验证 首先,因为开启和关闭应用的消息灰度功能后都需要重启节点才能生效,所以首先我们需要重启一下 springclouda 和 springcloudc 应用,重启的方式可以在控制台上选择重新部署,或者直接使用 kubectl 命令删除现有的 pod。 然后,继续使用 yaml 文件的方式在 Kubernetes 集群中部署新版本的 springcloudagray、springcloudbgray 和 springcloudcgray apiVersion: apps/v1 kind: Deployment metadata: name: springcloudagray spec: selector: matchLabels: app: springcloudagray template: metadata: annotations: alicloud.service.tag: gray msePilotCreateAppName: springclouda labels: app: springcloudagray spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre image: registry.cnshanghai.aliyuncs.com/yizhan/springclouda:1.0.0 imagePullPolicy: Always name: springcloudagray ports: containerPort: 20001 livenessProbe: tcpSocket: port: 20001 initialDelaySeconds: 10 periodSeconds: 30 apiVersion: apps/v1 kind: Deployment metadata: name: springcloudbgray spec: selector: matchLabels: app: springcloudbgray template: metadata: annotations: alicloud.service.tag: gray msePilotCreateAppName: springcloudb labels: app: springcloudbgray spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre image: registry.cnshanghai.aliyuncs.com/yizhan/springcloudb:1.0.0 imagePullPolicy: Always name: springcloudbgray ports: containerPort: 20002 livenessProbe: tcpSocket: port: 20002 initialDelaySeconds: 10 periodSeconds: 30 apiVersion: apps/v1 kind: Deployment metadata: name: springcloudcgray spec: selector: matchLabels: app: springcloudcgray template: metadata: annotations: alicloud.service.tag: gray msePilotCreateAppName: springcloudc labels: app: springcloudcgray spec: containers: env: name: JAVA_HOME value: /usr/lib/jvm/java1.8openjdk/jre image: registry.cnshanghai.aliyuncs.com/yizhan/springcloudc:1.0.0 imagePullPolicy: Always name: springcloudcgray ports: containerPort: 20003 livenessProbe: tcpSocket: port: 20003 initialDelaySeconds: 10 periodSeconds: 30 部署完成之后,我们引入流量,并进行验证 1. 登录 MSE 治理中心控制台[4],选择应用列表。 2. 单击应用 springclouda 应用详情菜单,此时可以看到,所有的流量请求都是去往 springclouda 应用未打标的版本,即稳定版本。 3. 点击页面下方的 标签路由中的添加按钮,为 springclouda 应用的 gray 版本设置灰度规则。 4. 发起流量调用,我们通过 zuulslb,分别发起流量调用,并查看灰度的情况。 我们通过 springclouda 和 springcloudagray 的日志去查看消息消费的情况。可以看到,消息的灰度功能已经生效, springcloudagray 这个环境,只会消费带有 gray 标的消息,springclouda 这个环境,只会消费未打标的流量生产出来的消息。 在截图中我们可以看见,springcloudagray 环境输出的日志  topic:TEST_MQ, producer: Cgray [10.25.0.102] , invoke result: Agray[10.25.0.101] Bgray[10.25.0.25] Cgray[10.25.0.102], springcloudagray 只会消费 Cgray 生产出来的消息,而且消费消息过程中发起的 Spring Cloud 调用,结果也是 Agray[10.25.0.101] Bgray[10.25.0.25] Cgray[10.25.0.102],即在灰度环境闭环。 而 springclouda 这个环境,输出的日志为 topic:TEST_MQ,producer:C[10.25.0.157],invoke result:A[10.25.0.100] B[10.25.0.152] C[10.25.0.157],只会消费 C 的基线环境生产出来的消息,且在这个过程中发起的 Spring Cloud 调用,也是在基线环境闭环。 步骤五:调整消息的标签过滤规则,并进行验证 因为考虑到实际场景中,springcloudc 应用和 springclouda  应用的所有者可能不是同一个人,不一定能够做到两者同时进行灰度发布同步的操作,所以在消息的灰度中,未打标环境默认的行为是消费所有消息。这样 springcloudc 在进行灰度发布的时候,可以不需要强制 springclouda 应用也一定要同时灰度发布,且使用相同的环境标。 springclouda 在消费时候,未打标环境的行为的选择权是交给 springclouda 的所有者,如果需要实现未打标环境不消费 cgray 生产出来的消息,只需要在控制台进行配置即可,配置之后实时生效。 1. 调整 springclouda 未打标环境的过滤规则。比如这里我们要选择未打标环境不再消费 gray 环境生产出来的消息,只需要在“未打标环境忽略的标签”里面选择 gray,然后点击确定即可。 2. 调整规则之后,规则是可以动态地生效,不需要进行重启的操作,我们直接查看 springclouda 的日志,验证规则调整生效。 从这个日志中,我们可以看到,此时基线环境可以同时消费 gray 和 基线环境生产出来的消息,而且在消费对应环境消息时产生的 Spring Cloud 调用分别路由到 gray 和 基线环境中。 操作总结 1. 全链路消息灰度的整个过程是不需要修改任何代码和配置的。 2. 目前仅支持 RocketMQ,Client 版本需要在 4.5.0 之后的版本。RocketMQ Server 端需要支持 SQL92 规则过滤,即开源 RocketMQ 需要配置 enablePropertyFilter=true,阿里云 RocketMQ 需要使用铂金版。 3. 开启消息灰度后,MSE Agent 会修改消息消费者的 group,如原来的消费 group 为 group1,环境标签为 gray,则 group 会被修改成 group1_gray,如果使用的是阿里云 RocketMQ,需要提前创建好修改后的 group。 4. 开启和关闭消息灰度后,应用需要重启才能生效;修改未打标环境忽略的标签功能可以动态生效,不需要重启。 相关链接 [1] MSE 微服务治理专业版: [2] Kubernetes 集群: [3] 容器服务控制台: [4] MSE 治理中心控制台 活动推荐 阿里云基于 Apache RocketMQ 构建的企业级产品消息队列RocketMQ 5.0版现开启活动: 1、新用户首次购买包年包月,即可享受全系列 85折优惠! 了解活动详情:
作者:亦盏
#行业实践 #功能特性