企业官网建设流程全解析

• 消息队列
• 后端
• 流处理

导读

Pulsar Functions 是 Apache Pulsar 内置的轻量级计算框架，它允许开发者直接以“函数”的形态编写消息处理逻辑，无需再部署独立的流处理系统（如 Storm、Flink）。本文将基于 Pulsar 2.2.1 版本文档与仓库源码，系统讲解 Pulsar Functions 的编程模型、两类原生/SDK 编程接口、命令行部署与配置方式、本地/集群两种运行模式，以及日志、用户配置、触发、处理保证、状态存储等核心机制，帮助你在 Pulsar 消息总线之上快速构建可投产的流式处理逻辑。

Pulsar Functions 是什么：核心目标与设计动机

Pulsar Functions 是“轻量级计算进程”，它做三件事：

• 从一个或多个 Pulsar topic消费消息；
• 对每条消息应用用户提供的处理逻辑；
• 把计算结果发布到另一个 topic。

Pulsar 官方文档将其定位为：让用户无需部署旁路系统（如 Apache Storm、Apache Heron、Apache Flink），即可在 Pulsar 消息体系内直接构建任意复杂度的处理逻辑——本质上，Pulsar Functions 是随消息系统一起交付的“现成计算基础设施”。

围绕这一核心目标，还派生出一系列子目标：

• 开发者生产力：支持语言原生函数（Language-native）与 Pulsar Functions SDK 两类写法，上手成本低；
• 易排查：函数可向日志 topic 输出日志、发布指标，便于定位问题；
• 运维简单：不需要维护额外的流处理集群，降低整体运维负担。

从设计渊源看，Pulsar Functions 借鉴了两类系统的思路：

• 流处理引擎（Apache Storm、Apache Heron、Apache Flink）的拓扑/算子思想；
• “Serverless / 函数即服务（FaaS）”云平台（AWS Lambda、Google Cloud Functions、Azure Functions）的按需计算模型。

可以把它概括为：“Lambda 风格的函数 + 以 Pulsar 作为消息总线”。每当你需要“对消息做点什么”时，都可以考虑用 Pulsar Functions 就地完成。

编程模型：输入、处理与三路输出

Pulsar Functions 的编程模型非常简洁：函数从一个或多个输入 topic接收消息，每当收到一条消息，函数可以：

• 对输入应用处理逻辑，并把结果写入Pulsar 中的输出 topic；
• 将结果写入Apache BookKeeper（状态存储）；
• 向日志 topic输出日志（常用于调试）；
• 累加counter 计数器。

从仓库源码看，这一模型由 Function 接口 直接落地：Function<I, O>的唯一抽象方法O process(I input, Context context) throws Exception会针对输入 topic 的每条消息被调用一次；接口还提供了default void initialize(Context context)（函数实例启动时初始化资源）与default void close()（实例停止时关闭资源）两个生命周期钩子，它们分别对应函数实例的“启动”与“停止”阶段，是编写有状态/有资源开销函数时的扩展点。

示例一：Exclamation 函数（Java 原生接口）

文档给出的 Java 原生接口（native interface）示例，用java.util.Function实现字符串追加感叹号：

import java.util.Function; public class ExclamationFunction implements Function<String, String> { @Override public String apply(String input) { return String.format("%s!", input); } }

Python 的等价实现（同样使用原生接口，直接定义一个process函数）：

def process(input): return "{0}!".format(input)

函数每次在输入 topic 收到消息时执行一次。例如函数监听tweet-streamtopic，那么每当有消息发布到该 topic，函数就会被触发运行一次。

示例二：词频统计（Word count）与状态计数器

经典的词频统计可以用 Pulsar Functions 很自然地实现：输入 topic 提供句子，函数对句子分词，并利用内置的计数器状态（由 BookKeeper 持久化）为每个单词累加计数。

使用 Java SDK 的写法如下（仓库中的完整实现见 WordCountFunction.java）：

package org.example.functions; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import java.util.Arrays; public class WordCountFunction implements Function<String, Void> { // 每有一条消息发布到输入 topic，该函数即被调用一次 @Override public Void process(String input, Context context) { Arrays.asList(input.split(" ")).forEach(word -> { String counterKey = word.toLowerCase(); context.incrCounter(counterKey, 1) }); return null; } }

注意：文档中的示例对每个单词做了toLowerCase()归一化（the与The视为同一个词）；仓库源码版本则直接以原始单词为 key（context.incrCounter(word, 1)），你可以按业务需要自由选择。context.incrCounter背后就是基于 Apache BookKeeper 的持久化计数器，因此即使函数实例重启，计数也不会丢失——这是它与普通局部变量计数器的本质区别。

用pulsar-admin在集群中部署该函数（集群运行模式，见下文）：

$ bin/pulsar-admin functions create \ --jar target/my-jar-with-dependencies.jar \ --classname org.example.functions.WordCountFunction \ --tenant public \ --namespace default \ --name word-count \ --inputs persistent://public/default/sentences \ --output persistent://public/default/count

示例三：基于内容的路由（Python）

内容路由是更复杂的典型场景：函数根据输入内容，把消息发布到不同的输出 topic。下图展示了“水果/蔬菜/都不是”三类路由：

Python SDK 实现如下：

from pulsar import Function class RoutingFunction(Function): def __init__(self): self.fruits_topic = "persistent://public/default/fruits" self.vegetables_topic = "persistent://public/default/vegetables" @staticmethod def is_fruit(item): return item in ["apple", "orange", "pear", "other fruits..."] @staticmethod def is_vegetable(item): return item in ["carrot", "lettuce", "radish", "other vegetables..."] def process(self, item, context): if self.is_fruit(item): context.publish(self.fruits_topic, item) elif self.is_vegetable(item): context.publish(self.vegetables_topic, item) else: warning = "The item {0} is neither a fruit nor a vegetable".format(item) context.get_logger().warn(warning)

这里用到的context.publish(topic, item)会经 SerDe 序列化后发布到指定 topic；context.get_logger().warn(...)则把警告写入函数配置的日志 topic。

命令行接口：pulsar-admin 与 pulsar-functions

Pulsar Functions 通过pulsar-adminCLI 工具管理，核心子命令是functions。一个在本地运行模式下启动函数的示例：

$ bin/pulsar-functions localrun \ --inputs persistent://public/default/test_src \ --output persistent://public/default/test_result \ --jar examples/api-examples.jar \ --classname org.apache.pulsar.functions.api.examples.ExclamationFunction

这里的org.apache.pulsar.functions.api.examples.ExclamationFunction是仓库中真实存在的示例类（见 ExclamationFunction.java），其process实现正是String.format("%s!", input)。相关pulsar-admin functions子命令还包括：create（集群模式部署）、trigger（触发函数）、list（列出函数）等。

全限定函数名（FQFN）

每个 Pulsar Function 都有一个全限定函数名（Fully Qualified Function Name，FQFN），由三部分组成：

tenant/namespace/name

例如public/default/word-count。FQFN 使得不同命名空间下可以存在同名函数而互不冲突。结合 Pulsar 的多租户模型（概念详见 concepts-multi-tenancy.md），FQFN 是函数在集群中的唯一寻址标识。

配置方式：CLI 参数与 YAML 文件

Pulsar Functions 支持两种配置方式，且可以混用：

1. 命令行参数：通过pulsar-admin functions接口传入；
2. YAML 配置文件：通过--function-config-file指定路径。

YAML 方式示例：

$ bin/pulsar-admin functions create \ --function-config-file ./my-function.yaml

对应的my-function.yaml：

name: my-function tenant: public namespace: default jar: ./target/my-functions.jar className: org.example.pulsar.functions.MyFunction inputs: - persistent://public/default/test_src output: persistent://public/default/test_result

你完全可以“混搭”：一部分属性走 CLI、另一部分走 YAML。此外，部署与管理工作负载时还支持默认参数机制——许多参数缺省时会被自动推导，例如：函数名缺省取类名（--classname org.example.MyFunction的函数名即为MyFunction）；tenant/namespace 缺省时从输入 topic 名推导；输出 topic 缺省为{input topic}-{function name}-output；处理保证缺省为ATLEAST_ONCE；服务地址缺省为pulsar://localhost:6650。

支持的编程语言与两类 API

Pulsar Functions 目前支持Java与Python两种语言。对应的 API 分两类：

Pulsar Functions API（SDK）

API 面向类型安全与 SerDe：

• 类型安全：函数既可以处理原始字节，也可以处理应用自定义的复杂类型；
• 基于 SerDe（序列化/反序列化）：多种内置类型开箱即用，也支持自定义 SerDe。

SDK 的核心类型定义在 pulsar-functions/api-java 目录下，包括Function<I, O>（核心函数接口）、Context（函数上下文）、SerDe（序列化接口）、Record（消息记录）、StateStore（状态存储接口）等。

函数上下文（Function context）

使用 SDK 创建的每个函数都能访问一个context 对象，它提供两类能力：

1. 函数自身信息：函数名、tenant、namespace、用户配置（user configuration）等；
2. 特殊功能：向指定日志 topic 输出日志、发布指标（metrics）。

从源码看，Context 接口 还提供：getInputTopics()（所有输入 topic）、getOutputTopic()（输出 topic）、getUserConfigMap()/getUserConfigValue(key)/getUserConfigValueOrDefault(key, defaultValue)（读取用户配置）、publish(topicName, object)（发布消息）、newOutputMessage(topicName, schema)（带 Schema 的发布）、getPulsarAdmin()（Pulsar Admin 客户端）等能力。

语言原生函数（Language-native functions）

Java 与 Python 都支持编写无任何依赖的“原生”函数（如前面的 Exclamation 示例）。其优点是零外部依赖、即写即用；缺点是无法访问 context，因而无法使用日志、用户配置、计数器等高级能力。需要这些能力时，请使用 SDK 版本。

SDK 示例：读取上下文的函数

Java（使用 SDK 访问上下文信息）：

import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import org.slf4j.Logger; public class ContextAwareFunction implements Function<String, Void> { @Override public Void process(String input, Context, context) { Logger LOG = context.getLogger(); String functionTenant = context.getTenant(); String functionNamespace = context.getNamespace(); String functionName = context.getName(); LOG.info("Function tenant/namespace/name: {}/{}/{}", functionTenant, functionNamespace, functionName); return null; } }

Python 等价实现：

from pulsar import Function class ContextAwareFunction(Function): def process(self, input, context): log = context.get_logger() function_tenant = context.get_function_tenant() function_namespace = context.get_function_namespace() function_name = context.get_function_name() log.info("Function tenant/namespace/name: {0}/{1}/{2}".format(function_tenant, function_namespace, function_name))

部署模式：本地运行与集群运行

Pulsar Functions 支持两种部署模式：

本地运行模式（Local run mode）

本地运行模式下，函数运行在执行命令的那台机器上（可能是笔记本、EC2 实例等）：

$ bin/pulsar-admin functions localrun \ --py myfunc.py \ --classname myfunc.SomeFunction \ --inputs persistent://public/default/input-1 \ --output persistent://public/default/output-1

默认情况下，函数会连接本机上的 Pulsar 集群（broker 服务地址pulsar://localhost:6650）。若要连接非本地的 Pulsar 集群，可用--broker-service-url指定：

$ bin/pulsar-admin functions localrun \ --broker-service-url pulsar://my-cluster-host:6650 \ # 其他函数参数

集群运行模式（Cluster run mode）

集群模式下，函数代码被上传到 Pulsar broker，并与 broker 一起运行（而不是跑在你的本地环境）。用create命令部署：

$ bin/pulsar-admin functions create \ --py myfunc.py \ --classname myfunc.SomeFunction \ --inputs persistent://public/default/input-1 \ --output persistent://public/default/output-1

该命令会把myfunc.py上传到 Pulsar，随后 Pulsar 依据代码启动一个（或按需多个）函数实例。

并行度（Parallelism）

集群模式下默认只启动1 个函数实例，但你可以通过并行度参数运行多个实例。例如创建并行度为 5（即 5 个实例）的函数：

$ bin/pulsar-admin functions create \ --name parallel-fun \ --tenant public \ --namespace default \ --py func.py \ --classname func.ParallelFunction \ --parallelism 5

并行度既可以在创建时指定，也可以事后更新已有的单实例函数。

函数实例资源（Function instance resources）

集群运行模式下，可以为每个函数实例分配资源：

示例：为函数分配 8 核、8 GB 内存、10 GB 磁盘：

$ bin/pulsar-admin functions create \ --jar target/my-functions.jar \ --classname org.example.functions.MyFunction \ --cpu 8 \ --ram 8589934592 \ --disk 10737418240

更多资源相关说明可参考 Deploying and Managing Pulsar Functions。

日志机制

使用 SDK 创建的 Pulsar Functions 可以把日志发送到函数配置中指定的日志 topic。例如下面的命令会把该函数的所有日志输出到persistent://public/default/my-func-1-log：

$ bin/pulsar-admin functions create \ --name my-func-1 \ --log-topic persistent://public/default/my-func-1-log \ # 其他配置

Java 函数中按输入内容在不同日志级别输出（仓库中有对应的 LoggingFunction.java 示例）：

public class LoggerFunction implements Function<String, Void> { @Override public Void process(String input, Context context) { Logger LOG = context.getLogger(); if (input.length() <= 100) { LOG.info("This string has a length of {}", input); } else { LOG.warn("This string is getting too long! It has {} characters", input); } } }

把日志集中到 Pulsar topic，意味着你可以用既有的消息消费链路（甚至另一个 Pulsar Function）统一收集和分析函数日志，而不必各自接入日志系统。

用户配置（User configuration）

Pulsar Functions 支持通过命令行传入任意 key-value（键和值都必须是字符串），这组键值对称为函数的用户配置，配置内容必须是 JSON 字符串。示例：

$ bin/pulsar-admin functions create \ --user-config '{"key-1":"value-1","key-2","value-2"}' \ # 其他配置

函数内通过 context 读取配置：

public class ConfigMapFunction implements Function<String, Void> { @Override public Void process(String input, Context context) { String val1 = context.getUserConfigValue("key1").get(); String val2 = context.getUserConfigValue("key2").get(); context.getLogger().info("The user-supplied values are {} and {}", val1, val2); return null; } }

从 Context 接口 的实现看，读取配置有三种方式：getUserConfigMap()（全量 Map）、getUserConfigValue(key)（返回Optional，需要.get()取值）以及getUserConfigValueOrDefault(key, defaultValue)（带默认值兜底）。这意味着无需修改函数代码，即可通过 CLI 动态改变函数行为（如阈值、目标 topic 名等），是一种典型的“配置与代码分离”实践。仓库示例 UserConfigFunction.java 展示了完整用法。

触发函数（Triggering）

集群模式下运行的函数可以被触发（trigger）——即通过 CLI 直接向函数传入一个值并拿到返回值，无需创建客户端、向输入 topic 发消息等繁琐步骤。触发特别适合测试与调试。

从机制上讲，触发一个函数与“向函数输入 topic 生产一条消息”并无本质区别；pulsar-admin functions trigger命令本质上只是“向函数发消息”的便捷机制，省去了使用pulsar-client或语言客户端库的步骤。

例如，下面这个反转字符串的 Python 原生函数：

def process(input): return input[::-1]

在集群中运行时可以这样触发：

$ bin/pulsar-admin functions trigger \ --tenant public \ --namespace default \ --name reverse-func \ --trigger-value "snoitcnuf raslup ot emoclew"

控制台输出应为welcome to pulsar functions。除了命令行传字符串，也可以用--triggerFile指定文件内容作为触发输入。

处理保证（Processing guarantees）

Pulsar Functions 提供三种消息语义，可应用到任意函数：

例如，以 effectively-once 语义在集群模式运行函数：

$ bin/pulsar-admin functions create \ --name my-effectively-once-function \ --processing-guarantees EFFECTIVELY_ONCE \ # 其他函数配置

三种语义的具体原理与取舍可进一步参考 Processing guarantees for Pulsar Functions 文档。

指标（Metrics）与状态存储（State storage）

• 指标：使用 Pulsar Functions SDK 的函数可以向 Pulsar 发布指标（如函数处理的消息总数、处理时长等），供监控系统采集分析。详见 Metrics for Pulsar Functions。
• 状态存储：Pulsar Functions 使用Apache BookKeeper作为状态存储接口（incrCounter之类的计数器即持久化于此）。所有 Pulsar 安装（包括本地 standalone 单机安装）都自带 BookKeeper bookie 部署，因此无需额外引入外部状态系统即可获得持久化、一致性的函数状态。更完整的 API 说明见 Pulsar Functions API，状态相关深入内容见 Pulsar Functions state storage。

快速上手路径与延伸阅读

如果你想立刻动手：

1. 阅读 Functions quick start，在一个 standalone 集群上跑通第一个函数；
2. 系统学习 Pulsar Functions API（Java/Python 的 SDK 与原生函数写法）；
3. 深入 Deploying and managing Pulsar Functions 掌握资源分配、触发、更新/删除等运维操作；
4. 仓库中 pulsar-functions/java-examples 提供了 38 个可直接阅读的示例函数（词频统计、日志、用户配置、发布、路由、窗口等），是学习不同用法的最佳范本；相关 Python 示例见 pulsar-functions/python-examples 目录（如exclamation_function.py、wordcount_function.py等）。

• 消息队列
• 后端
• 流处理