Apache Pulsar Functions 状态存储（State Storage）实战指南：基于 BookKeeper Table Service 的函数状态管理

消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载Pulsar Functions 自 2.1.0 起集成了 Apache BookKeeper 的 Table Service表格服务为无状态函数补齐了有状态能力函数可以把计数器counter或任意键值对key/value持久化到分布式存储中并在函数实例重启、迁移后依然可靠保留。本文以 version-2.2.1/functions-state.md 文档为骨架结合仓库源码逐层讲解 Pulsar Functions 的 State API、底层实现原理、状态查询命令与完整示例读完即可在自己的 Java 函数中直接落地状态存储。什么是 Pulsar Functions 状态存储Pulsar Functions 提供轻量级计算能力但其核心业务场景如窗口聚合、计数、去重、累加器往往需要跨消息、跨实例维护状态。为此从 Pulsar 2.1.0 开始Pulsar 将 Apache BookKeeper 的 Table Service 作为函数状态的底层存储函数通过 Pulsar Functions 的 State API 将状态写入 BookKeeper 的表格服务例如一个典型的WordCount函数可以把单词计数counters持久化到 BookKeeper Table Service 中从而实现即使函数实例重启计数也不会丢失。从源码结构看这一能力被组织在两层API 层pulsar-functions/api-java向开发者暴露统一的状态操作接口即Context对象上的 State API实现层pulsar-functions/instanceBKStateStoreImpl等实现类把 API 调用翻译为 BookKeeper Table Service 的底层 KV 操作。架构与实现原理State API 如何落到 BookKeeper理解状态存储最快的方式是直接看实现类。在 BKStateStoreImpl.java 中每个函数实例对应一个 BookKeeper 的TableByteBuf, ByteBuf表格服务句柄并持有该函数的三元组标识private final TableByteBuf, ByteBuf table; private final String tenant; private final String namespace; private final String name;状态存储的完整标识fqsnfully qualified state store name由FunctionCommon.getFullyQualifiedName(tenant, namespace, name)生成也就是每个函数在租户/命名空间/函数名维度上天然隔离不同函数之间的状态互不干扰。各 State API 与底层操作的对应关系一目了然State API底层 BookKeeper Table 操作说明incrCounter(key, amount)table.increment(key, amount)分布式原子自增getCounter(key)table.getNumber(key)读取计数器当前值putState(key, value)table.put(key, value)写入任意字节值getState(key)table.get(key)读取任意字节值deleteState(key)table.delete(key)删除键仓库实现中提供实现类还处理了几个容易被忽略的细节putAsync写入前会把ByteBuffer的position重置为 0。如果用户通过ByteBuffer.allocate(4).putInt(count)这类方式构造 buffer写完后 position 会停在末尾若不重置则 Table Service 将写不到任何数据getAsync返回时同样把结果 buffer 的 position 重置到开头避免用户在自己的函数代码里被迫手动rewind()getAsync在读取完 NettyByteBuf后通过ReferenceCountUtil.safeRelease及时释放引用防止内存泄漏。状态存储的 SPI 定义在 StateStore.java标注为Public、Evolving提供tenant()、namespace()、name()、fqsn()、init(StateStoreContext)与close()。在此基础上API 层又细分出两类语义接口CounterStateStore.java仅暴露计数器四件套incrCounter/incrCounterAsync/getCounter/getCounterAsyncByteBufferStateStore.java暴露通用 KV 操作put/putAsync/delete/deleteAsync/get/getAsync。实际的存储实例由 StateStoreProvider 与 BKStateStoreProviderImpl.java 创建后者从配置中读取stateStorageServiceUrl通过StorageClientSettings.newBuilder().serviceUri(stateStorageServiceUrl)构建存储客户端再按(tenant, namespace, name)创建或复用对应的状态表从而将函数实例与 BookKeeper Table Service 连接起来。Java State API 详解在 Java SDK 函数中全部 State API 都挂在 Context 对象上接口定义见 BaseContext.java分为计数器 API与通用键值 API两类且每个操作都提供同步与异步两个版本。计数器 APIincrCounter / getCounterincrCounter将指定key的计数器按amount递增/** * Increment the builtin distributed counter referred by key * param key The name of the key * param amount The amount to be incremented */ void incrCounter(String key, long amount);incrCounterAsync是异步版本立即返回CompletableFutureVoid不等待递增操作完成/** * Increment the builtin distributed counter referred by key * but dont wait for the completion of the increment operation * * param key The name of the key * param amount The amount to be incremented */ CompletableFutureVoid incrCounterAsync(String key, long amount);getCounter与getCounterAsync用于读取由incrCounter/incrCounterAsync维护的计数器值/** * Retrieve the counter value for the key. * * param key name of the key * return the amount of the counter value for this key */ long getCounter(String key); /** * Retrieve the counter value for the key, but dont wait * for the operation to be completed * * param key name of the key * return the amount of the counter value for this key */ CompletableFutureLong getCounterAsync(String key);通用键值 APIputState / getState除计数器外Pulsar Functions 还暴露通用键值 API允许函数存储任意字节数据/** * Update the state value for the key. * * param key name of the key * param value state value of the key */ void putState(String key, ByteBuffer value); /** * Update the state value for the key, but dont wait for the operation to be completed * * param key name of the key * param value state value of the key */ CompletableFutureVoid putStateAsync(String key, ByteBuffer value); /** * Retrieve the state value for the key. * * param key name of the key * return the state value for the key. */ ByteBuffer getState(String key); /** * Retrieve the state value for the key, but dont wait for the operation to be completed * * param key name of the key * return the state value for the key. */ CompletableFutureByteBuffer getStateAsync(String key);从 BaseContext.java 的接口定义看仓库中还进一步提供了deleteState(key)与deleteStateAsync(key)以及更灵活的getStateStore(tenant, namespace, name)通用状态存储访问入口可供需要多状态存储场景的开发者使用。Python SDK 的状态支持情况需要特别说明在 2.2.1 版本的时间节点上状态存储暂不支持 Python SDK文档明确标注 State currently is not supported at Python SDK因此上述 API 仅适用于 Java SDK 函数。启用状态存储的配置状态存储默认通过 Functions Worker 连接 BookKeeper Table Service。在 functions_worker.yml 的 State Management 小节中可以看到该配置项默认被注释即不显式配置时由集群初始化决定######################## # State Management ######################## # the service url points to bookkeeper table service # stateStorageServiceUrl: bk://localhost:4181stateStorageServiceUrl指向 BookKeeper Table Service 的地址格式为bk://host:port。在单机standalone模式下通常对应本地 BookKeeper 的 Table Service 端口该 URL 会被 BKStateStoreProviderImpl.java 读取并作为StorageClientSettings.serviceUri构建存储客户端。因此要使函数状态写入生效Functions Worker 必须能访问到该 BookKeeper 服务。用 pulsar-admin 查询与写入状态除了在函数代码里通过 State API 读写状态Pulsar 还提供 CLI 命令便于运维与调试时直接查看某个函数的状态。querystate查询函数状态文档给出的查询命令如下$ bin/pulsar-admin functions querystate \ --tenant tenant \ --namespace namespace \ --name function-name \ --state-storage-url bookkeeper-service-url \ --key state-key \ [---watch]--tenant、--namespace、--name定位目标函数--state-storage-urlBookKeeper Table Service 地址--key要查询的状态键--watch若指定CLI 将持续监听该键的值变化并每秒刷新一次直至手动中断。在 CmdFunctions.java 中可以看到该命令的当前实现StateGetter通过getAdmin().functions().getFunctionState(tenant, namespace, functionName, key)拉取状态并以 JSON 格式打印--key简写-k与--watch简写-w是命令行参数。其中--watch模式下若状态尚不存在返回 404程序会打印错误信息后继续每秒重试直到键值出现——这对调试状态是否已写入非常有用。putstate向函数写入状态仓库中还提供了对应的写入命令putstate该命令由 CmdFunctions.java 中StatePutter实现注册为jcommander.addCommand(putstate, getStatePutter())$ bin/pulsar-admin functions putstate \ --tenant tenant \ --namespace namespace \ --name function-name \ --state {key:state-key,stringValue:value}--state参数接收一个 JSON 序列化的FunctionState对象必需项内部通过ObjectMapperFactory反序列化后调用putFunctionState写入。与querystate配合可以在不触发函数的情况下手工注入或修正某个键的状态值。实战示例WordCountFunction文档与仓库共同推荐的入门示例是 WordCount 函数完整源码见 WordCountFunction.javapublic class WordCountFunction implements FunctionString, Void { Override public Void process(String input, Context context) { Arrays.asList(input.split(\\s)).forEach(word - context.incrCounter(word, 1)); return null; } }该函数的逻辑非常简单直接将收到的输入String按正则切分成多个单词2.2.1 版文档中给出的示例使用\\.切分句子当前仓库中的示例演进为按空白符\s切分实践中可根据业务需要选择分隔符对每个word通过context.incrCounter(word, 1)将其对应计数器递增 1。由于incrCounter底层是 BookKeeper Table Service 的原子自增操作见 BKStateStoreImpl.java 中table.increment(key, amount)的调用即使函数被并发执行、实例发生重启或迁移单词计数也能保持一致且持久。之后便可以用上文中的querystate命令按单词键实时查看累计计数。测试用例行为验证仓库为状态存储提供了完整的单元测试见 BKStateStoreImplTest.java其中覆盖了incrCounter/getCounter验证递增后计数器的读取值并校验底层table.getNumber的调用put/get验证字节值写入与读取并校验table.put/table.get的调用次数getAsync验证异步读取路径包括键不存在时返回null的边界行为。这些测试直接对BKStateStoreImpl进行 mock 驱动是理解状态存储行为契约尤其是 ByteBuffer 位置处理与异步语义的最佳参考。小结与使用建议能力定位Pulsar Functions 状态存储把 BookKeeper Table Service 封装为函数级分布式状态覆盖计数器incrCounter/getCounter与通用键值putState/getState两类语义全部 API 均有异步版本适用场景需要跨消息累积状态的函数计数、统计、聚合、去重以及需要把中间结果持久化的有状态函数配置与运维通过functions_worker.yml的stateStorageServiceUrl指向 BookKeeper Table Service日常调试用pulsar-admin functions querystate --key key配合--watch实时观察需要手工修正状态时用pulsar-admin functions putstate版本限制截至 2.2.1 文档版本Python SDK 尚不支持状态存储状态 API 仅面向 Java SDK 函数实现细节状态按(tenant, namespace, name)完全隔离ByteBuffer的读写位置由实现层统一重置用户无需手动处理接口标注为Evolving后续版本 API 可能演进升级 Pulsar 时建议关注 BaseContext.java 与 StateStore.java 的变更记录。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar Functions 状态存储State Storage开发指南基于 BookKeeper Table Service 的有状态函数实战Apache Pulsar Functions 状态存储State Storage开发指南基于 BookKeeper Table Service 的有状态消息队列后端流处理Apache Pulsar Functions 状态存储State Storage开发指南基于 BookKeeper Table Service 的分布式状态管理Apache Pulsar Functions 状态存储State Storage开发指南基于 BookKeeper Table Service 的分布式消息队列后端流处理Apache Pulsar Functions 状态存储机制基于 BookKeeper Table Service 的 State API 全解析Apache Pulsar Functions 状态存储机制基于 BookKeeper Table Service 的 State API 全解析 本篇技术指消息队列后端流处理上一篇前端监控告警阈值设置终极指南动态与静态阈值完全解析下一篇如何编译Project Mu中的Rust代码构建流程与文档规范完整教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考