消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本篇技术指南围绕 Apache Pulsar 2.3.0 版本的官方文档 Pulsar Functions State Storage (Developer Preview) 展开系统讲解 Pulsar Functions 的分布式状态存储State StorageState API 的完整方法清单incrCounter/getCounter/putState/getState及其异步版本、pulsar-admin functions querystate状态查询命令、以及经典 WordCount 示例的编写方式。读完后你将能够理解 Pulsar Functions 如何利用 Apache BookKeeper Table Service 持久化函数状态并结合本仓库源码看清每个 API 的底层调用链。状态存储的整体设计自 Pulsar 2.1.0 起Pulsar 将函数状态State的存储集成到 Apache BookKeeper 的 Table Service 上。例如一个WordCount函数可以把每个单词的计数器counters通过 State API 写入 BookKeeper 的 table service从而在函数实例之间、以及函数重启之后都保持一致的计数结果。这一设计带来的关键能力分布式一致性状态由 BookKeeper 的复制日志保证持久化函数实例无状态化可随意扩缩容内置计数器原语State API 直接暴露incrCounter底层对应 BookKeeper table 的原子自增操作天然适合词频统计、流量计数等场景通用 KV 存储除计数器外还暴露通用的 key/value 接口putState/getState值为ByteBuffer可存储任意序列化后的状态数据外部可查询Pulsar 同时提供pulsar-admin functions querystateCLI可在函数运行期间直接查看某个 key 的当前状态值。Java State API 完整清单当前 Pulsar Functions 通过Context对象向 Java SDK 函数暴露以下状态操作 API。这些 API 定义在 BaseContext 接口中Context接口继承自它每个操作都提供同步与异步两种形式。incrCounter增量更新内置分布式计数器/** * 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);应用可以通过incrCounter将给定key对应的计数器增加指定的amount。incrCounterAsync异步自增/** * 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);incrCounterAsync与同步版本逻辑相同但不等待自增操作完成返回一个CompletableFutureVoid适合高吞吐场景下避免阻塞处理线程。getCounter读取计数器/** * 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);应用通过getCounter读取此前被incrCounter修改过的计数器值。getCounterAsync异步读取计数器/** * 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 the key */ CompletableFutureLong getCounterAsync(String key);getCounterAsync异步获取给定key的计数器值返回CompletableFutureLong。putState写入通用 KV 状态/** * 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);除计数器 API 外Pulsar 还暴露通用的 key/value 接口供函数存储任意二进制状态值为ByteBuffer。putStateAsync异步写入 KV 状态/** * 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);getState读取 KV 状态/** * Retrieve the state value for the key. * * param key name of the key * return the state value for the key. */ ByteBuffer getState(String key);getStateAsync异步读取 KV 状态/** * 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);Python SDK 的支持情况按照该版本文档的说明Python SDK 函数当时尚不支持 State。如果你需要状态能力应优先使用 Java SDK 编写函数。底层实现BKStateStoreImpl 与 BookKeeper Table Service结合本仓库源码可以看到 State API 的完整落地路径。Context的状态调用最终委托给DefaultStateStore的实现类 BKStateStoreImpl其核心是一个 BookKeeper 的TableByteBuf, ByteBuf句柄public class BKStateStoreImpl implements DefaultStateStore { private final String tenant; private final String namespace; private final String name; private final String fqsn; private final TableByteBuf, ByteBuf table; Override public CompletableFutureVoid incrCounterAsync(String key, long amount) { // TODO: this can be optimized with a batch operation. return table.increment( Unpooled.wrappedBuffer(key.getBytes(UTF_8)), amount); } ... }几个值得关注的实现细节计数器是自增原语incrCounterAsync直接调用table.increment(key, amount)即 BookKeeper Table Service 的原子自增保证多实例并发incrCounter时不会丢失更新同步版本incrCounter则是阻塞等待该 Future 完成失败时抛出RuntimeException计数器读取getCounterAsync调用table.getNumber(key)把存储中的值解释为数字返回KV 写入与 ByteBuffer position 陷阱putAsync中会显式执行value.position(0)——源码注释解释了原因如果用户通过ByteBuffer.allocate(4).putInt(count)之类的方式构造缓冲区position 已经位于末尾若不回卷到起点将什么都写不进 table service。这是一个非常容易被忽略的坑KV 读取时的缓冲回卷与内存释放getAsync在把ByteBuf的字节读入新的ByteBuffer后会执行result.position(0)回卷起点并用finally块中的ReferenceCountUtil.safeRelease(data)释放 Netty 引用的内存状态命名空间状态存储按tenant/namespace/name三元组定位fqsn FunctionCommon.getFullyQualifiedName(...)与函数的租户/命名空间体系一致。从源码结构看BookKeeper 之外的状态实现如 PulsarMetadataStateStoreImpl也实现了同一套DefaultStateStore契约意味着 State API 的抽象层是存储可插拔的。配置Worker 侧的 State Storage 参数要让状态存储生效Functions Worker 需要配置 BookKeeper Table Service 的连接信息。在仓库中的 conf/functions_worker.yml 可以看到对应配置段######################## # State Management ######################## # the service url points to bookkeeper table service # stateStorageServiceUrl: bk://localhost:4181在 WorkerConfig 中状态管理相关字段有两个配置项说明默认值stateStorageServiceUrl状态存储BookKeeper Table Service的服务地址无默认值需显式配置如bk://localhost:4181stateStorageProviderImplementation状态存储实现类org.apache.pulsar.functions.instance.state.BKStateStoreProviderImpl即默认状态下 Pulsar Functions 使用 BookKeeper 实现作为 State 提供者只需把stateStorageServiceUrl指向可用的 BookKeeper 集群即可启用状态存储能力。使用 CLI 查询函数状态Query State除了函数内部的 State APIPulsar 还提供 CLI 命令用于从外部直接查询函数写入的状态。命令形式为$ bin/pulsar-admin functions querystate \ --tenant tenant \ --namespace namespace \ --name function-name \ --state-storage-url bookkeeper-service-url \ --key state-key \ [---watch]如果指定--watchCLI 会持续监视给定state-key的值。对照 CmdFunctions 中的StateGetter命令实现querystate子命令可以确认各参数的真实行为--key简写-k必填状态键名缺失时抛出ParameterException(State key needs to be specified)--watch简写-w开启后进入do...while(watch)轮询循环每次调用getAdmin().functions().getFunctionState(tenant, namespace, functionName, key)并将FunctionState以 JSONGson 美化输出打印到标准输出循环间隔为Thread.sleep(1000)即每秒一次当状态键不存在HTTP 404且处于 watch 模式时只会向标准错误打印消息而继续等待否则直接抛出异常--tenant/--namespace继承自FunctionCommand缺省时分别回退到public租户和default命名空间。从源码结构看该 CLI 还注册了putstate子命令StatePutter可通过--state传入 JSON 表示的FunctionState直接写入状态方便调试时预置或修正状态值。实战示例WordCount 函数使用 StateWordCountFunction 是展示如何轻松使用 Pulsar Functions 状态存储的经典示例核心逻辑仅两行public 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; } }这个WordCount函数的逻辑简单直接函数先将接收到的String按空白字符切分成多个单词当前仓库版本使用正则\\s分割早期文档示例中曾写作按\\.分割对每个word调用context.incrCounter(word, 1)将对应计数器自增 1。由于计数器存储在 BookKeeper Table Service 中即使该函数部署了多个实例、或实例发生重启词频统计依然全局一致——函数本身完全无状态。部署后可用前文的pulsar-admin functions querystate --key word验证某个单词的累计计数。验证集成测试与单元测试仓库中提供了针对状态存储的完整测试覆盖可作为行为依据PulsarStateTesttests/integration模块端到端验证提交 WordCount 类函数后通过管理 API 查询函数状态的行为BKStateStoreImplTest针对BKStateStoreImpl的单元测试覆盖 BookKeeper table 上的增/读/写操作ContextImplTest验证Context实现类对 State API 的委托行为。小结Pulsar Functions 的状态存储机制由三层构成面向函数开发者的ContextState API同步/异步双通道中间由DefaultStateStore契约解耦底层由BKStateStoreImpl落到 BookKeeper Table Service 的原子自增与 KV 操作上运维侧则通过functions_worker.yml的stateStorageServiceUrl启用、通过pulsar-admin functions querystate可加--watch持续监视进行外部状态查询。理解了这套机制你就可以把任意有状态逻辑词频统计、会话状态、去重标记等安全地构建在 Pulsar Functions 之上。版本说明本文基于 2.3.0 版本文档整理状态存储在该版本中标注为 Developer Preview文中源码证据来自当前仓库含后续版本演进如deleteState、putstate等新增能力。赞分享消息队列后端流处理【免费下载链接】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 PIP-312 深度解析基于 StateStoreProvider 解耦 Pulsar Functions 状态存储与 BookKeeperApache Pulsar PIP 312 深度解析基于 StateStoreProvider 解耦 Pulsar Functions 状态存储与 BookK消息队列后端上一篇如何使用Chart.js浏览器扩展xhubGitHub页面图表增强完整指南下一篇Pixelle-Video 上手指南3 步跑通 AI 短视频生成流水线创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
