关于阿里云:消息服务-Serverless-函数计算如何助力企业降本提效

作者：柳下

背景介绍

音讯队列服务（下文均以 Message Service 命名）作为云计算 PaaS 畛域的基础设施之一，其 高并发、削峰填谷的个性 愈发受到开发者关注。Message Service 对上承接音讯生产者服务的申请，对下连贯消费者服务。提到生产：那就不得不引入两个问题？

1. 如何以低成本、高吞吐、低延时的形式将音讯数据从 Message Service 输送给上游生产服务？
2. 如何疾速构建免运维、按需弹性伸缩算力的音讯生产服务？

明天就来聊聊如何在阿里云上基于 Serverless 计算服务 + Message Service 构建这样一套零碎。

名词解释

函数计算（Function Compute）

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

官网文档：

https://help.aliyun.com/produ…

连接器（Connector）

Connector 实现了大量数据的导入和导出。例如将 KAFKA topic 中数据导出到 stdout，或将本地文件中数据导入到 RocketMQ。Connector 简化了数据在不同零碎间复制和传输的复杂度，本文探讨的音讯服务和计算服务的连贯同样依赖 Connector 实现。

事件总线（EventBridge）

事件总线是 Connector 的产品化服务，反对阿里云服务、自定义利用、SaaS 利用等以标准化、中心化的形式接入，并可能以标准化协定在这些利用之间路由事件，帮忙您轻松构建松耦合、分布式的事件驱动架构，更多产品细节可浏览官网文档 [2 ]。

官网文档：

https://help.aliyun.com/produ…

架构演进

传统的数据生产架构如下图左：

1. 数据源将产生的数据写入到音讯零碎；
2. 开发者借助 Message Service 提供的 OpenAPI/SDK 或 Proxy 服务客户端从 Message Service 读取数据；
3. 依据音讯数据处理业务逻辑，也就是咱们所谓的生产音讯，将音讯生产的业务结写入到指标服务；

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

• 如何并发平安的从 Message Service 读取数据？
• 数据生产能力小于生产能力时，如何疾速晋升生产吞吐？
• 指标服务资源成为瓶颈时，如何疾速扩容？当流量波峰过后，面对闲暇的机器老本，您又如何解决？
• 如何保障生产实时性、程序性？
• 如何实现容错、缓存、降级、限流等高可用爱护伎俩？
• 如何监控链路状态或异样？
• ……

面对下面多个琐碎又简单的问题，置信总有几个会击中您的痛点。为了同时解决提到的所有问题，阿里云开发 Connector Service（如上图右）买通 Message Service 和 Serverless 计算服务的数据链路，您只需申明上游的音讯服务实例和上游的生产算子，便可一键部署上线，连接器同时提供了丰盛的流计算框架具备的数据处理能力和监控能力，总结如下：

1. Transform：以 UDF 形式自定义数据荡涤逻辑，同时反对 JsonPath 语法简略提取数据；
2. Filter：缩小无用音讯的后续解决，提供多种过滤匹配规定，如：前后缀匹配、数值匹配、IP 地址匹配等；
3. Window：提供窗口能力，可依照音讯数量和间隔时间对音讯做聚合推送。可晋升音讯解决吞吐，升高音讯解决老本；
4. Real Time：从 Message Service 拉取音讯到推送指标服务延时毫秒级别；
5. 自定义并发生产能力：并发平安的生产音讯，晋升吞吐能力；
6. 弹性计算资源：上游计算服务依据负载主动扩缩容，无需关怀服务器资源水位问题；
7. Monitoring + Logging + Tracing：提供了丰盛的监控指标和日志剖析助力开发者监控零碎状态、定位异样；
8. 齐备的异样保障机制：自定义重试策略 + 容错机制 + 死信队列 + 限流 + 反压；

为让大家对性能有更深刻的理解，上面咱们具体介绍各个性能的好处和利用场景。

降本提效性能

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 ]

官网文档：

https://help.aliyun.com/produ…

客户案例

客户需要

某广告平台每天将浏览的用户信息（个人信息、工夫、登录设施等）投递至 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 控制台

https://kafka.console.aliyun….

kakfa 疾速入门

https://help.aliyun.com/docum…

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 的成熟更加升高了简单零碎的开发门槛，让您真正实现端到端全链路深度上云。