消息服务 + Serverless 函数计算如何助力企业降本提效？-六虎

作者：柳下

布景介绍

音讯行列服务（下文均以 Message Service 命名）作为云核算 PaaS 领域的基础设施之一，其高并发、削峰填谷的特性愈发遭到开发者重视。Message Service 对上承接音讯生产者服务的恳求，对下衔接消费者服务。说到消费：那就不得不引入两个问题？

怎么以低本钱、高吞吐、低延时的方式将音讯数据从 Message Service 输送给下流消费服务？
怎么快速构建免运维、按需弹性弹性算力的音讯消费服务？

今日就来聊聊怎么在阿里云上根据 Serverless 核算服务 + Message Service 构建这样一套体系。

名词解释

函数核算（Function Compute）

阿里云函数核算是事情驱动的全托管 Serverless 核算服务。经过函数核算，您无需办理服务器等基础设施，只需编写代码并上传。函数核算会为您准备好核算资源，以弹性、可靠的方式运行您的代码，更多产品细节可阅览官方文档 [ 1] 。

官方文档：

help.aliyun.com/product/509…

衔接器（Connector）

Connector 完结了很多数据的导入和导出。例如将 KAFKA topic 中数据导出到 stdout，或将本地文件中数据导入到 RocketMQ。Connector 简化了数据在不同体系间仿制和传输的杂乱度，本文讨论的音讯服务和核算服务的衔接同样依赖 Connector 完结。

事情总线（EventBridge）

事情总线是 Connector 的产品化服务，支撑阿里云服务、自定义使用、SaaS 使用等以标准化、中心化的方式接入，并能够以标准化协议在这些使用之间路由事情，帮助您轻松构建松耦合、分布式的事情驱动架构，更多产品细节可阅览官方文档 [2 ] 。

官方文档：

help.aliyun.com/product/161…

架构演进

传统的数据消费架构如下图左：

数据源将发生的数据写入到音讯体系；
开发者凭借 Message Service 供给的 OpenAPI/SDK 或 Proxy 服务客户端从 Message Service 读取数据；
依据音讯数据处理事务逻辑，也便是咱们所谓的消费音讯，将音讯消费的事务结写入到方针服务；

如此架构开发者会面临以下几个问题：

怎么并发安全的从 Message Service 读取数据？
数据消费才能小于生产才能时，怎么快速提高消费吞吐？
方针服务资源成为瓶颈时，怎么快速扩容？当流量波峰往后，面临闲暇的机器本钱，您又怎么处理？
怎么确保消费实时性、顺序性？
怎么完结容错、缓存、降级、限流等高可用维护手法？
怎么监控链路状况或反常？
……

面临上面多个琐碎又杂乱的问题，相信总有几个会击中您的痛点。为了一起处理说到的一切问题，阿里云开发 Connector Service（如上图右）打通 Message Service 和 Serverless 核算服务的数据链路，您只需声明上游的音讯服务实例和下流的消费算子，便可一键布置上线，衔接器一起供给了丰厚的流核算框架具有的数据处理才能和监控才能，总结如下：

Transform：以 UDF 方式自定义数据清洗逻辑，一起支撑 JsonPath 语法简略提取数据；
Filter：削减无用音讯的后续处理，供给多种过滤匹配规矩，如：前后缀匹配、数值匹配、IP 地址匹配等；
Window：供给窗口才能，可依照音讯数量和距离时刻对音讯做聚合推送。可提高音讯处理吞吐，降低音讯处理本钱；
Real Time：从 Message Service 拉取音讯到推送方针服务延时毫秒级别；
自定义并发消费才能：并发安全的消费音讯，提高吞吐才能；
弹性核算资源：下流核算服务依据负载自动扩缩容，无需关怀服务器资源水位问题；
Monitoring + Logging + Tracing：供给了丰厚的监控指标和日志剖析助力开发者监控体系状况、定位反常；
完备的反常保障机制：自定义重试战略 + 容错机制 + 死信行列 + 限流 + 反压；

为让我们对功用有更深入的了解，下面咱们具体介绍各个功用的好处和使用场景。

降本提效功用

Window

在大规划数据场景中，One Message Per Request 早已无法满足开发者需求。Window 本质是供给了一种音讯攒批处理的才能，Connector 在产品层面供给两个可分配参数：

批量推送条数：单次聚合的最大音讯条数，当积压的音讯数量到达设定值时才会将音讯推送到下流。
批量推送距离：体系每到距离时刻点会将积压的音讯聚合后发给下流，假如设置 0 秒表示无等候时刻，接纳即投递。

两个参数结合使用可极大提高数据传输功率，从而提高数据吞吐，一起能够解锁多种用户场景，例：

流形式实时消费：将推送距离设为 0s，推送条数设置最大值，这样能够确保从上游拉到的数据实时推送到下流方针服务。
恳求稀少且延时不敏感场景下，期望音讯被攒批处理，能够接受消费滞后但不期望滞后时刻过长：假如仅设置批量推送条数一个参数，则或许在低谷期由于音讯稀少长期无法到达预设的攒批条数而滞后过久，此刻可引入批量推送距离参数处理此问题。

Transform

音讯消费离不开数据处理，所谓数据处理，便是经过某个进程将原始数据转为方针数据，转化的进程即为 transform。一般原始数据是一个大而全的信息集合，而方针数据仅仅一个结构化的子集，要害在于怎么嵌入数据的清洗和提取才能。对此 Connector 供给了多种转化才能：

Template：关于原数据和方针数据都是确定结构的数据，且数据提取组装规矩简略，能够凭借模版完结 transform，模版一起支撑 JsonPath 数据提取规矩，如下图：

UDF（User Define Function 用户自定义函数）：对原数据结构杂乱，且数据转化进程杂乱的场景，能够凭借 UDF 完结。UDF 形式中，服务供给方仅约定了函数的入参协议、参数的数据结构，至于函数中怎么对数据做清洗？返回的数据结构怎么？全部交由开发者完结，极大提高了音讯处理的灵活度，一个简略的 UDF demo 如下：

# -*- coding: utf-8 -*-
# handle_message 为函数执行进口
# 服务供给方约定了入参 event 和 context 的数据格式
# 只需从 event 中解析音讯体并做处理即可
def handle_message(event, context):
    try:
        new_message = transform(event)
    except Exception as e:
        raise e
    return new_message
def transform(old_message):
  # 自定义对数据的清洗和处理逻辑，并返回处理后的音讯
  return new_message

Filter

Filter 削减无用音讯的后续处理，提高音讯处理的功率，尤其和 Serverless 核算结合时，可削减调用次数，例如以下场景：

对敏感字、不合法文字、要害字进行过滤；
对某些具有攻击性的 IP 进行音讯拦截；
……

为掩盖足够多的事务场景，Connector 供给了前缀匹配、后缀匹配、数值匹配、IP 地址匹配等多种匹配形式，您能够依据事务需求挑选适合的形式。

Real Time

在流核算场景中，低延时消费是开发者比较重视的一个问题，Connector 在供给批处理才能的一起也兼顾了流处理场景，当时刻攒批窗口设置为 0 时，体系将演变为实时消费行为。

自定义并发消费才能

以 KAFKA 为例，当 KAFKA 数据量增大时，用户一般凭借 Topic Partition 的水平扩展才能提高投递和消费的速率，跟着 Topic Partition 分区数的不断添加，Consumer 端仍沿袭单线程消费一切 partition 数据的方案一定会遇到瓶颈，从而导致音讯积压。为了处理此问题，Connector 开放了自定义并发消费线程数装备，您能够指定多个 consumer threads，多个 consumer threads 会均分 kafka 的多个 partition，防止音讯积压问题。当 Topic Partition 数量和 Consumer 线程数持平时可到达最大吞吐（如下图 3），一起可做到 Partition 粒度保序。

高可用维护战略

重试：由于网络反常、体系 crash 等原因导致音讯消费反常时，体系会按装备的 Retry Policy 进行重试，现在支撑退避重试、指数衰减重试；
死信行列：当音讯超越重试次数后仍未消费成功时，就变成了死信音讯，假如不期望死信音讯被丢掉，能够装备死信行列，一切的死信音讯会被体系投递到死信行列中，现在体系支撑 KAFKA、RocketMQ、MNS 作为死信行列的方针端；
容错战略：当音讯消费发生错误时，体系供给以下两种处理方式：

- 答应容错：答应反常容错，当反常发生时不会堵塞执行，超出重试战略后会依据装备将音讯投递至死信行列或直接丢掉，继续消费下一条音讯；
- 禁止容错：不答应错误，当反常发生并超越重试战略装备时会堵塞执行；
反压：当体系接纳音讯的速率远高于它的处理速率时，出于对体系的维护会触发反压机制，防止体系崩溃，反压在体系中体现在两方面：

- 从上游拉音讯的速率大于下流消费速率：积压的音讯逐渐增多，假如不操控上游的拉取速率，会导致 Connector 内存不足造成 OOM；
- 下流方针服务限流：当方针服务受衔接数、网络带宽等资源约束无法服务更多恳求时，会返回给 Connector 很多限流错误，假如 Connector 不操控音讯消费速率，或许引发体系雪崩；

针对上面两种场景，体系均经过技术手法做了维护，技术细节暂不描述。

弹性核算资源

Connector 打通了音讯服务和 Serverless 函数核算服务，您或许会忧虑一个问题：函数核算服务的算力能否实时适配上游音讯规划的不断增加？答案是能够的。函数核算作为 Serverless 核算服务，底层的核算资源能够做到毫秒级弹性，不论您的 consumer 端并发消费才能怎么调整，投递音讯的频率有多高，函数核算均可在 quota 范围内快速弹性核算实例。

核算实例 Quota 是函数核算出于对事务方服务维护设置的最大并发运行实例数，假如实际事务规划大于此默认值，能够给函数核算团队提工单调高此值。

Connector 结构

Connector 定义了数据从哪里仿制到哪里，经过和谐调度一系列 Task 完结数据的传输作业，Task 依据职责不同可划分为以下几类：

Poller Task：从上游音讯服务中拉取音讯；
Transform Task：对音讯做清洗、加工、过滤、聚合等操作；
Sink Task：将音讯推送到下流服务；

Task 均可水平扩展，并发消费上游多 partition 数据，且并发将音讯投递到下流处理。

当前 Connector 依赖阿里云 EventBridge 完结，更多才能可参阅官方文档 [3 ]

官方文档：

help.aliyun.com/product/161…

客户案例

客户需求

某广告平台每天将阅读的用户信息（个人信息、时刻、登录设备等）投递至 kafka 中，从事务视点投递的数据格式并不完全相同，客户需将不同格式的数据清洗为相同格式的数据，并将清洗后的数据投递到 ClickHouse 服务，跟着用户事务日益增加，估计未来几个月有几倍增加，且客户对实时性和本钱都有要求，总结客户的几点要害需求如下：

具有数据清洗才能；
低本钱；
体系不受事务增加要素影响；

处理方案

函数核算刚好能够完美处理上述问题，下面结合如下数据链路介绍怎么处理客户的几个需求：

怎么完结数据清洗？
Transform Task 中供给了 Data Cleaning 功用，客户能够以 UDF 方式自定义数据清洗逻辑，平台规定了入参协议，出参能够为任意格式的清洗后数据；
怎么做到低本钱？
整条链路主要费用源于函数核算的核算资源耗费和调用次数，可经过以下两个手法降低本钱：

- Window：将多条音讯聚合为一条批量音讯发送至函数核算，削减调用次数，防止重复执行公共核算逻辑；
- Filter：削减无用音讯的后续处理，削减调用函数核算的次数；
怎么确保体系不受事务增加要素影响？

经过下图可发现，kafka topic 的 partition 分区数、Poller 数量、Sink Task 的 worker 数量、函数核算的核算实例数都可完结任意水平扩展，且均可经过装备调整，因而当客户预判到事务增加时，只需修正相应的装备项即可完结水平扩容。

客户事务现状

现在客户已将事务全量搬迁到函数核算，搬迁后的几个月内仅经过简略修正扩容装备轻松应对事务规划的数倍增加。

最佳实践

下文经过演示一个将 kafka 数据导入到函数核算的 demo，快速建立一套音讯消费体系：

1、创立上游服务

登录kafka 操控台 [4 ] 创立 kakfa 实例，并在该实例下创立 topic 和 groupID，能够参阅kakfa 快速入门 [5 ] 快速完结此操作。

kafka操控台

kafka.console.aliyun.com/

kakfa 快速入门

help.aliyun.com/document_de…

2、创立下流服务 + 装备数据处理规矩

a.创立函数核算的服务，并为服务命名，如下图：

b.在创立的服务下创立一个函数，函数是执行代码的最小单元，如下图：

c.在创立函数页面，为函数命名，并点击触发器装备，其中触发器类型挑选 kakfa，将 step1 创立的资源（kakfa 实例、Topic、Group ID ）填写到下图中，其他值可使用默认值。

d.(可选) 如需求验证攒批功用，可点击批量推送，并装备批量推送条数和批量推送距离，此 demo 设置批量推送条数为 2 条，批量推送距离为 10s，如下图：

e.上面流程完结后点击确定即布置成功。

3、编写函数，函数内的逻辑为输出接纳到的音讯数量和音讯内容：

# -*- coding: utf-8 -*-
import logging
import json
def handler(event, context):
  evt = json.loads(event)
  logger = logging.getLogger()
  logger.info(len(evt)) // 输出音讯列表的长度
  logger.info(evt)。// 输出音讯内容
  return 'succ'

4、测验验证

a.到 kafka 操控台的 topic 中快速发送 3 条音讯，如下图：

b.预期函数核算会收到 2 次恳求，第 1 次恳求由于触发推送条数条件包括 2 条音讯，第 2 次恳求在等候 10s 后触发推送距离条件包括 1 条音讯，如下图：

c.可经过函数日志查看一切恳求日志，能够发现总共接纳到 3 条音讯，如下图：

总结& 展望

根据 Serverless 函数核算，您能够快速建立一套安全可靠的数据消费体系，总结体系优势如下：

降本

- Filter：削减无效的音讯处理和对函数核算的调用；
- Window：供给音讯攒批处理才能，帮助更好处理一些非实时和离散场景下的音讯，也削减了对函数核算的调用次数；
- 按需付费：核算资源按需付费的特性防止了波峰波谷场景下为峰值预留机器发生的无用开销；
- 持续降价：函数核算在 11 月份下调全地域全计费项价格，下调幅度达 12%-47%，并对内存和 cpu 做精细化计费；
提效

- 研发功率：Transform、UDF、Template、JsonPath 等才能解锁更多事务场景，防止二次开发，助您快速构建体系，未来也会内嵌更丰厚的算子，乃至能够编列算子；
- 数据剖析功率：供给数值检索、可视化剖析等才能，您能够经过简略的引导式交互，即可快速完结根据事情的流式查询与剖析；
- 问题排查功率：体系供给丰厚的可观测才能，如事情轨道、事情大盘等助您对事务进行监控和整体状况剖析，未来也会从指标探索、运维监控、毛病定位等多个维度完善才能，完结更全面的体系可观测性；
- 运维功率：Serverless 核算实例毫秒级自动弹性弹性的特性让你彻底摆脱资源运维的担负；

跟着云核算逐渐走向全面 Serverless 化，Message Service 和 Serverless 核算的衔接会愈加严密，如今 Connector 的老练愈加降低了杂乱体系的开发门槛，让您真正完结端到端全链路深度上云。

消息服务 + Serverless 函数计算如何助力企业降本提效？