FastStream Confluent Kafka 发布消息全指南broker.publish、Publisher 对象与装饰器三种实践【免费下载链接】faststreamAsynchronous Python framework for event-driven services. A thin client for Kafka, RabbitMQ, NATS, Redis and MQTT with full access to native broker features, plus AsyncAPI docs, in-memory tests and observability out of the box.项目地址: https://gitcode.com/GitHub_Trending/fa/faststream本文围绕 FastStream 的 Confluent Kafka 适配器faststream.confluent.KafkaBroker展开完整讲解消息发布的三种官方用法直接调用broker.publish(...)、创建可复用的 Publisher 对象、以及使用 Publisher 装饰器串联订阅与发布管道。文章不仅继承官方文档的全部配置与代码示例还结合仓库源码与测试用例深入剖析publish方法签名、生产者级参数acks、partitioner、linger_ms等、AsyncAPI 文档化机制与内存测试方式帮助你按需选择最合适的发布方案并理解其底层原理。FastStream 与 Confluent Kafka统一的发布模型FastStream 是面向事件驱动服务的异步 Python 框架对 Kafka含 Confluent 客户端、RabbitMQ、NATS、Redis、MQTT 等 Broker 提供统一的 API。其中faststream.confluent模块基于confluent-kafka-python实现其KafkaBroker支持所有常规的发布用例即框架通用的broker.publish、broker.publisher装饰器、Publisher 对象等用法且无需任何改动即可直接使用。官方文档 Publisher/index.md 指出如果你希望进一步定制发布逻辑则需要关注KafkaBroker特有的一些参数。本指南将三种发布方式逐一展开并给出对应的源码依据与可运行的完整示例。准备工作创建 KafkaBroker 实例无论使用哪种发布方式第一步都是创建 Broker 实例。KafkaBroker的构造函数位于 faststream/confluent/broker/broker.py最简用法只需传入 Kafka 地址from faststream.confluent import KafkaBroker broker KafkaBroker(localhost:9092)该构造函数将参数分为几组其中与发布Producer直接相关的关键参数如下参数默认值说明bootstrap_serverslocalhosthost[:port]字符串或列表默认端口 9092用于引导获取集群元数据client_idSERVICE_NAME客户端标识会随每次请求发给服务端便于定位服务端日志acks未设置等价于1生产者要求 Leader 收到多少确认才认为请求完成0不等待任何确认1仅 Leader 写入本地日志即确认all或-1等待全部 ISR 副本确认最强可靠性compression_typeNone压缩类型gzip、snappy、lz4、zstdpartitionerconsistent_random决定每条消息分配到哪个分区的可调用对象默认对非None的 key 使用与 Java 客户端相同的 murmur2 哈希保证同 key 消息落到同一分区key 为None时随机选择分区max_request_size1024 * 1024单次请求也即单条记录的最大字节数linger_ms0批量发送前的等待延迟用于在中等负载下聚合更多记录、减少请求次数enable_idempotenceFalse开启后保证每条消息恰好写入一份开启时acks会被强制设为alltransactional_idNone生产者事务 ID设置后支持事务性消息transaction_timeout_ms60 * 1000事务超时时间毫秒此外构造函数还支持request_timeout_ms、retry_backoff_ms、metadata_max_age_ms、connections_max_idle_ms、allow_auto_create_topics等连接层参数以及graceful_timeout、ack_policy、logger、log_level、middlewares、routers、security、AsyncAPI 相关参数specification_url、protocol、description、tags等。从源码可以看到这些参数最终会封装进ConfluentFastConfig见 broker.py再交给底层的confluent_kafka生产者与消费者使用。因此在创建KafkaBroker时统一配置acks、compression_type等参数会对所有发布行为生效。方式一直接调用 broker.publish 发布消息基础发布KafkaBroker通过统一的publish方法来自 producer 对象发送消息这是最基础、最直接的发布方式。你可以使用 Python 原生类型或pydantic.BaseModel定义消息内容并通过 topic 名称指定发送目标。下面的完整示例来自仓库 docs/docs_src/confluent/raw_publish/example.py它演示了「创建 Broker 实例 → 定义消息模型与订阅函数 → 在测试中直接发布」的完整流程import pytest from pydantic import BaseModel, Field, NonNegativeFloat from faststream import FastStream, Logger from faststream.confluent import KafkaBroker, TestKafkaBroker broker KafkaBroker(localhost:9092) app FastStream(broker) class Data(BaseModel): data: NonNegativeFloat Field( ..., examples[0.5], descriptionFloat data example, ) broker.subscriber(input_data) async def handle_data(msg: Data, logger: Logger) - None: logger.info(handle_data(msg%s), msg) pytest.mark.asyncio async def test_raw_publish(): async with TestKafkaBroker(broker): msg Data(data0.5) await broker.publish( msg, topicinput_data, ) handle_data.mock.assert_called_once_with(dict(msg))按照官方文档的步骤拆解创建 KafkaBroker 实例broker KafkaBroker(localhost:9092)并包装为FastStream(broker)使用publish方法发布消息await broker.publish(msg, topicinput_data)其中msg是一个 Pydantic 模型实例topic指定目标 topic 名称。publish 方法签名与参数详解KafkaBroker.publish的完整签名定义在 faststream/confluent/broker/broker.pyasync def publish( self, message: SendableMessage, topic: str, *, key: bytes | str | None None, partition: int | None None, timestamp_ms: int | None None, headers: dict[str, str] | None None, correlation_id: str | None None, reply_to: str , no_confirm: bool False, ) - asyncio.Future[Message | None] | Message | None:各参数含义与源码 docstring 一致message消息体可以是任意 JSON 可序列化对象Python 原生类型、Pydantic 模型或原始bytestopic消息发布的 topic 名称key消息 key用于分区选择。当partition为None且 partitioner 保持默认时相同 key 的消息会被投递到同一分区key 为None时随机选择分区。key 必须是bytes或能通过配置的 key 序列化器转换为bytespartition指定目标分区不设置时由partitioner决定timestamp_ms消息时间戳毫秒headers消息头用于存放元信息。content-type与correlation_id由框架自动设置这里可以补充自定义头correlation_id手动指定消息关联 ID。若不指定框架会调用id_generator生成默认为基于 UUID4 的gen_cor_id用于跨服务追踪消息处理链路reply_to响应消息要发送到的 topic 名称配合 RPC / Request-Reply 场景使用no_confirmFalse时等待 Kafka 的发布确认后再返回返回Message | NoneTrue时不等待确认直接返回asyncio.Future。从源码实现看publish会构造一个KafkaPublishCommand并通过_basic_publish(cmd, producerself.config.producer)提交给底层 producer 发送见 broker.py。_basic_publish定义于 faststream/_internal/broker/pub_base.py它会按逆序包装 broker 中间件middleware最终调用 producer 的publish——这意味着 broker 级发布中间件对该路径同样生效。序列化规则与自动头FastStream 允许发布任意 JSON 可序列化消息或原始字节并自动设置必要头详见 getting-started 发布基础correlation_id默认每次publish(...)/request(...)未显式指定时生成随机 UUID4可通过构造函数传入id_generator替换例如改用按创建时间可字典序排序的 ULIDcontent-typeFastStream 服务的语义化头帮助框架依据该头快速选择序列化器。可选值为text/plain、application/json、空值配合字节内容。非原始字节消息推荐统一使用application/json完全省略头也可以但会使序列化略慢。基础发布的局限这种直接发布方式有一个显著限制你的发布行为不会出现在 AsyncAPI 文档中源码 broker.py 的 docstring 也明确说明这是“非 AsyncAPI 文档化”的发布路径建议仅在其它框架应用或偶尔发消息时使用。如果只是偶尔发送一次性消息这种方式完全可以接受但如果要构建完整服务官方文档建议改用下面两种方式。方式二创建 Publisher 对象可复用、可文档化创建 Publisher 对象是解决「文档化」问题的第一步将broker.publisher(topic)的返回值保存下来之后反复调用该对象的publish方法。这些对象会被 FastStream 解析并写入服务的 AsyncAPI 文档。完整示例见 docs/docs_src/confluent/publisher_object/example.pyimport pytest from pydantic import BaseModel, Field, NonNegativeFloat from faststream import FastStream, Logger from faststream._internal._compat import model_to_json from faststream.confluent import KafkaBroker, TestKafkaBroker broker KafkaBroker(localhost:9092) app FastStream(broker) class Data(BaseModel): data: NonNegativeFloat Field( ..., examples[0.5], descriptionFloat data example, ) prepared_publisher broker.publisher(input_data) broker.subscriber(input_data) async def handle_data(msg: Data, logger: Logger) - None: logger.info(handle_data(msg%s), msg) pytest.mark.asyncio async def test_prepared_publish(): async with TestKafkaBroker(broker): msg Data(data0.5) await prepared_publisher.publish( model_to_json(msg), headers{content-type: application/json}, ) handle_data.mock.assert_called_once_with(dict(msg))步骤拆解创建 KafkaBroker 实例broker KafkaBroker(localhost:9092)创建 Publisher 实例prepared_publisher broker.publisher(input_data)通过预置的 Publisher 发布消息await prepared_publisher.publish(model_to_json(msg), headers{content-type: application/json})。当 Broker 被包装进FastStream对象后这个 publisher 就会导出到 AsyncAPI 文档中参见下文「AsyncAPI 文档化机制」一节。publisher 注册方法的参数broker.publisher(...)的完整签名定义于 faststream/confluent/broker/registrator.pydef publisher( self, topic: Union[str, Topic], *, key: bytes | str | None None, partition: int | None None, headers: dict[str, str] | None None, reply_to: str , batch: bool False, persistent: bool True, title: str | None None, description: str | None None, schema: Any | None None, include_in_schema: bool True, autoflush: bool False, ) - Union[BatchPublisher, DefaultPublisher]:要点说明topic接受字符串或Topic对象但FastStream 永远不会为 publisher 创建 topic因此Topic的创建设置会被忽略只有名称有意义见 faststream/confluent/publisher/factory.py 的注释key、partition、headers、reply_to会被固化为该 publisher 的默认值后续调用publish时可再覆盖headers中content-type与correlation_id无论如何都会被框架自动设置batchTrue时返回BatchPublisher支持一次发布多条消息autoflushTrue时每次发布后都会调用 producer 的flush()实现见 factory.pytitle、description、schema、include_in_schema用于控制 AsyncAPI 文档中的描述信息schema应为 Python 原生类型注解或pydantic.BaseModel。Publisher 对象的发布方法DefaultPublisher.publish定义于 faststream/confluent/publisher/usecase.py签名与broker.publish基本一致但topic默认为空字符串此时使用创建时固化的 topic其余参数均可按次覆盖。同一文件中的BatchPublisher.publish则接收*messages可变参数用于批量发送。方式三使用 Publisher 装饰器串联处理管道装饰器是 FastStream 推荐的第三种也是最适合快速开发应用的方式它同时提供 AsyncAPI 表示并构造一个「输入 输出」的 DataPipeline 单元。官方文档特别强调两点装饰器顺序不影响功能broker.subscriber(...)与broker.publisher(...)的叠加顺序无关紧要装饰器只能应用于已被broker.subscriber(...)装饰的函数该方法依赖处理函数的返回值类型注解框架依据返回类型注解来正确解释并序列化函数返回值后再发送因此返回类型注解必须准确。先看完整应用来自 docs/docs_src/confluent/publish_example/app.pyfrom pydantic import BaseModel, Field, NonNegativeFloat from faststream import FastStream from faststream.confluent import KafkaBroker class Data(BaseModel): data: NonNegativeFloat Field( ..., examples[0.5], descriptionFloat data example, ) broker KafkaBroker(localhost:9092) app FastStream(broker) to_output_data broker.publisher(output_data) to_output_data broker.subscriber(input_data) async def on_input_data(msg: Data) - Data: return Data(datamsg.data 1.0)按官方文档的四步拆解初始化 KafkaBroker 实例broker KafkaBroker(localhost:9092)包含必要的 Kafka 地址配置准备 Publisher 对象留作装饰器to_output_data broker.publisher(output_data)编写处理逻辑定义一个消费指定格式入站消息、并向指定 topic 产出响应的函数async def on_input_data(msg: Data) - Data: return Data(datamsg.data 1.0)装饰处理函数用broker.subscriber(input_data)与to_output_data同时装饰。应用启动后每当订阅的 topic 出现新消息处理函数即被调用其返回值会被发布到 Publisher 装饰器指定的 topicoutput_datato_output_data broker.subscriber(input_data) async def on_input_data(msg: Data) - Data: return Data(datamsg.data 1.0)对应测试见 tests/docs/confluent/publish_example/test_app.py它使用TestKafkaBroker验证了整条链路向input_data发布Data(data0.2)后on_input_data.mock被调用一次且收到{data: 0.2}同时to_output_data.mock被调用一次且收到{data: 1.2}—— 精确印证了「订阅 → 处理 → 自动发布返回值」的数据流。装饰器模式的底层机制当to_output_data装饰的函数被 subscriber 触发后框架会调用 publisher 的_publish方法见 faststream/confluent/publisher/usecase.py将命令的destination指向该 publisher 的 topic合并 publisher 固化的headers且overrideFalse即不覆盖处理流程中已有的头补齐reply_to、partition、key等默认值最终经由_basic_publish走中间件链并调用 producer 发送。这一设计使得「每个 subscriber 处理函数 返回值发布」成为一个结构化的数据处理单元既清晰又便于在 AsyncAPI 中呈现。进阶指定分区键发布Key 分区语义key参数在 Kafka 分区模型中至关重要默认consistent_randompartitioner 会对非Nonekey 做 murmur2 哈希保证同 key 消息落到同一分区从而保证同一业务键如用户 ID的消息有序。仓库中的 docs/docs_src/confluent/publish_with_partition_key/app.py 演示了如何通过 Context 读取入站消息的 key并在发布时显式指定 keyto_output_data broker.publisher(output_data) broker.subscriber(input_data) async def on_input_data( msg: Data, logger: Logger, key: bytes Context(message.raw_message.key), ) - None: logger.info(on_input_data(msg%s), msg) await to_output_data.publish(Data(datamsg.data 1.0), keybkey) broker.subscriber(output_data) async def on_output_data( msg: Data, logger: Logger, key: bytes Context(message.raw_message.key), ) - None: logger.info(on_output_data(msg%s), msg)这里通过Context(message.raw_message.key)注入原始 Kafka 消息的 key并在publish(...)调用中传keybkey即可在消费侧同样用 Context 读取 key实现按 key 对齐的分区读写。发布路径的底层调用链综合源码KafkaBroker的发布路径可以归纳为入口KafkaBroker.publish(...)broker.py或DefaultPublisher.publish(...)usecase.py命令构造将消息、topic、key、partition、headers、correlation_id 等封装为KafkaPublishCommandcorrelation_id未指定时调用self.config.id_generator()生成中间件包装_basic_publishpub_base.py按逆序将 broker 中间件包装到 producer 的publish调用上底层发送由AsyncConfluentFastProducerImpl执行实际的confluent_kafka发送并按no_confirm决定是否等待 Kafka 确认。KafkaBroker还额外提供publish_batch(*messages, topic...)与request(message, topic, ...)两个发布族方法见 broker.py前者批量发送多条消息后者执行 Request-Reply 并等待响应消息。它们同样封装为KafkaPublishCommand并复用中间件与 producer 链路。AsyncAPI 文档化机制三种方式在 AsyncAPI 文档中的表现截然不同broker.publish(...)不进入文档broker.publisher(topic)创建的 Publisher 对象 / 装饰器当 Broker 被包装进FastStream对象后会导出到 AsyncAPI 文档。其实现位于 faststream/confluent/publisher/specification.pyKafkaPublisherSpecification会为每个 publisher 生成一个规范条目默认名称形如{topic}:Publisher例如output_data:Publisher包含address发布目标 topicoperation.message.payload依据 publisher 的schema未显式指定时从处理函数返回类型解析生成的 payload 定义bindings.kafkaKafka Channel Binding携带topic信息。这解释了为什么官方文档强调构建完整服务时应使用 Publisher 对象或装饰器以便让发布接口也纳入 AsyncAPI 契约。内存测试TestKafkaBroker上面三个示例都使用了TestKafkaBroker进行内存测试无需真实 Kafka 集群async with TestKafkaBroker(broker): await broker.publish(msg, topicinput_data) handle_data.mock.assert_called_once_with(dict(msg))TestKafkaBroker会替换底层连接将订阅处理函数替换为可断言的mock对象同时发布消息也会被记录因此你可以像断言订阅一样断言发布例如上文测试中的to_output_data.mock.assert_called_once_with(dict(Data(data1.2)))。这使得三种发布方式都具备开箱即用的可测试性。对应测试文件可见 tests/docs/confluent/raw_publish/test_raw_publish.py 与 tests/docs/confluent/publisher_object/test_publisher_object.py。三种发布方式的选择建议方式语法AsyncAPI 文档化适用场景broker.publish(...)await broker.publish(msg, topic...)否一次性消息、集成其它框架应用、临时发送Publisher 对象p broker.publisher(t); await p.publish(msg)是需要复用发布目标、在服务内部多处发布Publisher 装饰器broker.publisher(t)叠加在 subscriber 函数上是快速构建「订阅-处理-再发布」管道自动发布函数返回值在发布参数定制方面KafkaBroker构造函数统一配置acks、compression_type、partitioner、max_request_size、linger_ms、enable_idempotence等生产者参数单次发布时则通过key、partition、timestamp_ms、headers、correlation_id、reply_to、no_confirm精细控制。结合 官方发布基础文档 中关于序列化与correlation_id的约定即可在真实项目中落地稳定、可追踪、可文档化的 Kafka 发布能力。【免费下载链接】faststreamAsynchronous Python framework for event-driven services. A thin client for Kafka, RabbitMQ, NATS, Redis and MQTT with full access to native broker features, plus AsyncAPI docs, in-memory tests and observability out of the box.项目地址: https://gitcode.com/GitHub_Trending/fa/faststream创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastStream Confluent Kafka 发布消息全指南：broker.publish、Publisher 对象与装饰器三种实践
FastStream Confluent Kafka 发布消息全指南broker.publish、Publisher 对象与装饰器三种实践【免费下载链接】faststreamAsynchronous Python framework for event-driven services. A thin client for Kafka, RabbitMQ, NATS, Redis and MQTT with full access to native broker features, plus AsyncAPI docs, in-memory tests and observability out of the box.项目地址: https://gitcode.com/GitHub_Trending/fa/faststream本文围绕 FastStream 的 Confluent Kafka 适配器faststream.confluent.KafkaBroker展开完整讲解消息发布的三种官方用法直接调用broker.publish(...)、创建可复用的 Publisher 对象、以及使用 Publisher 装饰器串联订阅与发布管道。文章不仅继承官方文档的全部配置与代码示例还结合仓库源码与测试用例深入剖析publish方法签名、生产者级参数acks、partitioner、linger_ms等、AsyncAPI 文档化机制与内存测试方式帮助你按需选择最合适的发布方案并理解其底层原理。FastStream 与 Confluent Kafka统一的发布模型FastStream 是面向事件驱动服务的异步 Python 框架对 Kafka含 Confluent 客户端、RabbitMQ、NATS、Redis、MQTT 等 Broker 提供统一的 API。其中faststream.confluent模块基于confluent-kafka-python实现其KafkaBroker支持所有常规的发布用例即框架通用的broker.publish、broker.publisher装饰器、Publisher 对象等用法且无需任何改动即可直接使用。官方文档 Publisher/index.md 指出如果你希望进一步定制发布逻辑则需要关注KafkaBroker特有的一些参数。本指南将三种发布方式逐一展开并给出对应的源码依据与可运行的完整示例。准备工作创建 KafkaBroker 实例无论使用哪种发布方式第一步都是创建 Broker 实例。KafkaBroker的构造函数位于 faststream/confluent/broker/broker.py最简用法只需传入 Kafka 地址from faststream.confluent import KafkaBroker broker KafkaBroker(localhost:9092)该构造函数将参数分为几组其中与发布Producer直接相关的关键参数如下参数默认值说明bootstrap_serverslocalhosthost[:port]字符串或列表默认端口 9092用于引导获取集群元数据client_idSERVICE_NAME客户端标识会随每次请求发给服务端便于定位服务端日志acks未设置等价于1生产者要求 Leader 收到多少确认才认为请求完成0不等待任何确认1仅 Leader 写入本地日志即确认all或-1等待全部 ISR 副本确认最强可靠性compression_typeNone压缩类型gzip、snappy、lz4、zstdpartitionerconsistent_random决定每条消息分配到哪个分区的可调用对象默认对非None的 key 使用与 Java 客户端相同的 murmur2 哈希保证同 key 消息落到同一分区key 为None时随机选择分区max_request_size1024 * 1024单次请求也即单条记录的最大字节数linger_ms0批量发送前的等待延迟用于在中等负载下聚合更多记录、减少请求次数enable_idempotenceFalse开启后保证每条消息恰好写入一份开启时acks会被强制设为alltransactional_idNone生产者事务 ID设置后支持事务性消息transaction_timeout_ms60 * 1000事务超时时间毫秒此外构造函数还支持request_timeout_ms、retry_backoff_ms、metadata_max_age_ms、connections_max_idle_ms、allow_auto_create_topics等连接层参数以及graceful_timeout、ack_policy、logger、log_level、middlewares、routers、security、AsyncAPI 相关参数specification_url、protocol、description、tags等。从源码可以看到这些参数最终会封装进ConfluentFastConfig见 broker.py再交给底层的confluent_kafka生产者与消费者使用。因此在创建KafkaBroker时统一配置acks、compression_type等参数会对所有发布行为生效。方式一直接调用 broker.publish 发布消息基础发布KafkaBroker通过统一的publish方法来自 producer 对象发送消息这是最基础、最直接的发布方式。你可以使用 Python 原生类型或pydantic.BaseModel定义消息内容并通过 topic 名称指定发送目标。下面的完整示例来自仓库 docs/docs_src/confluent/raw_publish/example.py它演示了「创建 Broker 实例 → 定义消息模型与订阅函数 → 在测试中直接发布」的完整流程import pytest from pydantic import BaseModel, Field, NonNegativeFloat from faststream import FastStream, Logger from faststream.confluent import KafkaBroker, TestKafkaBroker broker KafkaBroker(localhost:9092) app FastStream(broker) class Data(BaseModel): data: NonNegativeFloat Field( ..., examples[0.5], descriptionFloat data example, ) broker.subscriber(input_data) async def handle_data(msg: Data, logger: Logger) - None: logger.info(handle_data(msg%s), msg) pytest.mark.asyncio async def test_raw_publish(): async with TestKafkaBroker(broker): msg Data(data0.5) await broker.publish( msg, topicinput_data, ) handle_data.mock.assert_called_once_with(dict(msg))按照官方文档的步骤拆解创建 KafkaBroker 实例broker KafkaBroker(localhost:9092)并包装为FastStream(broker)使用publish方法发布消息await broker.publish(msg, topicinput_data)其中msg是一个 Pydantic 模型实例topic指定目标 topic 名称。publish 方法签名与参数详解KafkaBroker.publish的完整签名定义在 faststream/confluent/broker/broker.pyasync def publish( self, message: SendableMessage, topic: str, *, key: bytes | str | None None, partition: int | None None, timestamp_ms: int | None None, headers: dict[str, str] | None None, correlation_id: str | None None, reply_to: str , no_confirm: bool False, ) - asyncio.Future[Message | None] | Message | None:各参数含义与源码 docstring 一致message消息体可以是任意 JSON 可序列化对象Python 原生类型、Pydantic 模型或原始bytestopic消息发布的 topic 名称key消息 key用于分区选择。当partition为None且 partitioner 保持默认时相同 key 的消息会被投递到同一分区key 为None时随机选择分区。key 必须是bytes或能通过配置的 key 序列化器转换为bytespartition指定目标分区不设置时由partitioner决定timestamp_ms消息时间戳毫秒headers消息头用于存放元信息。content-type与correlation_id由框架自动设置这里可以补充自定义头correlation_id手动指定消息关联 ID。若不指定框架会调用id_generator生成默认为基于 UUID4 的gen_cor_id用于跨服务追踪消息处理链路reply_to响应消息要发送到的 topic 名称配合 RPC / Request-Reply 场景使用no_confirmFalse时等待 Kafka 的发布确认后再返回返回Message | NoneTrue时不等待确认直接返回asyncio.Future。从源码实现看publish会构造一个KafkaPublishCommand并通过_basic_publish(cmd, producerself.config.producer)提交给底层 producer 发送见 broker.py。_basic_publish定义于 faststream/_internal/broker/pub_base.py它会按逆序包装 broker 中间件middleware最终调用 producer 的publish——这意味着 broker 级发布中间件对该路径同样生效。序列化规则与自动头FastStream 允许发布任意 JSON 可序列化消息或原始字节并自动设置必要头详见 getting-started 发布基础correlation_id默认每次publish(...)/request(...)未显式指定时生成随机 UUID4可通过构造函数传入id_generator替换例如改用按创建时间可字典序排序的 ULIDcontent-typeFastStream 服务的语义化头帮助框架依据该头快速选择序列化器。可选值为text/plain、application/json、空值配合字节内容。非原始字节消息推荐统一使用application/json完全省略头也可以但会使序列化略慢。基础发布的局限这种直接发布方式有一个显著限制你的发布行为不会出现在 AsyncAPI 文档中源码 broker.py 的 docstring 也明确说明这是“非 AsyncAPI 文档化”的发布路径建议仅在其它框架应用或偶尔发消息时使用。如果只是偶尔发送一次性消息这种方式完全可以接受但如果要构建完整服务官方文档建议改用下面两种方式。方式二创建 Publisher 对象可复用、可文档化创建 Publisher 对象是解决「文档化」问题的第一步将broker.publisher(topic)的返回值保存下来之后反复调用该对象的publish方法。这些对象会被 FastStream 解析并写入服务的 AsyncAPI 文档。完整示例见 docs/docs_src/confluent/publisher_object/example.pyimport pytest from pydantic import BaseModel, Field, NonNegativeFloat from faststream import FastStream, Logger from faststream._internal._compat import model_to_json from faststream.confluent import KafkaBroker, TestKafkaBroker broker KafkaBroker(localhost:9092) app FastStream(broker) class Data(BaseModel): data: NonNegativeFloat Field( ..., examples[0.5], descriptionFloat data example, ) prepared_publisher broker.publisher(input_data) broker.subscriber(input_data) async def handle_data(msg: Data, logger: Logger) - None: logger.info(handle_data(msg%s), msg) pytest.mark.asyncio async def test_prepared_publish(): async with TestKafkaBroker(broker): msg Data(data0.5) await prepared_publisher.publish( model_to_json(msg), headers{content-type: application/json}, ) handle_data.mock.assert_called_once_with(dict(msg))步骤拆解创建 KafkaBroker 实例broker KafkaBroker(localhost:9092)创建 Publisher 实例prepared_publisher broker.publisher(input_data)通过预置的 Publisher 发布消息await prepared_publisher.publish(model_to_json(msg), headers{content-type: application/json})。当 Broker 被包装进FastStream对象后这个 publisher 就会导出到 AsyncAPI 文档中参见下文「AsyncAPI 文档化机制」一节。publisher 注册方法的参数broker.publisher(...)的完整签名定义于 faststream/confluent/broker/registrator.pydef publisher( self, topic: Union[str, Topic], *, key: bytes | str | None None, partition: int | None None, headers: dict[str, str] | None None, reply_to: str , batch: bool False, persistent: bool True, title: str | None None, description: str | None None, schema: Any | None None, include_in_schema: bool True, autoflush: bool False, ) - Union[BatchPublisher, DefaultPublisher]:要点说明topic接受字符串或Topic对象但FastStream 永远不会为 publisher 创建 topic因此Topic的创建设置会被忽略只有名称有意义见 faststream/confluent/publisher/factory.py 的注释key、partition、headers、reply_to会被固化为该 publisher 的默认值后续调用publish时可再覆盖headers中content-type与correlation_id无论如何都会被框架自动设置batchTrue时返回BatchPublisher支持一次发布多条消息autoflushTrue时每次发布后都会调用 producer 的flush()实现见 factory.pytitle、description、schema、include_in_schema用于控制 AsyncAPI 文档中的描述信息schema应为 Python 原生类型注解或pydantic.BaseModel。Publisher 对象的发布方法DefaultPublisher.publish定义于 faststream/confluent/publisher/usecase.py签名与broker.publish基本一致但topic默认为空字符串此时使用创建时固化的 topic其余参数均可按次覆盖。同一文件中的BatchPublisher.publish则接收*messages可变参数用于批量发送。方式三使用 Publisher 装饰器串联处理管道装饰器是 FastStream 推荐的第三种也是最适合快速开发应用的方式它同时提供 AsyncAPI 表示并构造一个「输入 输出」的 DataPipeline 单元。官方文档特别强调两点装饰器顺序不影响功能broker.subscriber(...)与broker.publisher(...)的叠加顺序无关紧要装饰器只能应用于已被broker.subscriber(...)装饰的函数该方法依赖处理函数的返回值类型注解框架依据返回类型注解来正确解释并序列化函数返回值后再发送因此返回类型注解必须准确。先看完整应用来自 docs/docs_src/confluent/publish_example/app.pyfrom pydantic import BaseModel, Field, NonNegativeFloat from faststream import FastStream from faststream.confluent import KafkaBroker class Data(BaseModel): data: NonNegativeFloat Field( ..., examples[0.5], descriptionFloat data example, ) broker KafkaBroker(localhost:9092) app FastStream(broker) to_output_data broker.publisher(output_data) to_output_data broker.subscriber(input_data) async def on_input_data(msg: Data) - Data: return Data(datamsg.data 1.0)按官方文档的四步拆解初始化 KafkaBroker 实例broker KafkaBroker(localhost:9092)包含必要的 Kafka 地址配置准备 Publisher 对象留作装饰器to_output_data broker.publisher(output_data)编写处理逻辑定义一个消费指定格式入站消息、并向指定 topic 产出响应的函数async def on_input_data(msg: Data) - Data: return Data(datamsg.data 1.0)装饰处理函数用broker.subscriber(input_data)与to_output_data同时装饰。应用启动后每当订阅的 topic 出现新消息处理函数即被调用其返回值会被发布到 Publisher 装饰器指定的 topicoutput_datato_output_data broker.subscriber(input_data) async def on_input_data(msg: Data) - Data: return Data(datamsg.data 1.0)对应测试见 tests/docs/confluent/publish_example/test_app.py它使用TestKafkaBroker验证了整条链路向input_data发布Data(data0.2)后on_input_data.mock被调用一次且收到{data: 0.2}同时to_output_data.mock被调用一次且收到{data: 1.2}—— 精确印证了「订阅 → 处理 → 自动发布返回值」的数据流。装饰器模式的底层机制当to_output_data装饰的函数被 subscriber 触发后框架会调用 publisher 的_publish方法见 faststream/confluent/publisher/usecase.py将命令的destination指向该 publisher 的 topic合并 publisher 固化的headers且overrideFalse即不覆盖处理流程中已有的头补齐reply_to、partition、key等默认值最终经由_basic_publish走中间件链并调用 producer 发送。这一设计使得「每个 subscriber 处理函数 返回值发布」成为一个结构化的数据处理单元既清晰又便于在 AsyncAPI 中呈现。进阶指定分区键发布Key 分区语义key参数在 Kafka 分区模型中至关重要默认consistent_randompartitioner 会对非Nonekey 做 murmur2 哈希保证同 key 消息落到同一分区从而保证同一业务键如用户 ID的消息有序。仓库中的 docs/docs_src/confluent/publish_with_partition_key/app.py 演示了如何通过 Context 读取入站消息的 key并在发布时显式指定 keyto_output_data broker.publisher(output_data) broker.subscriber(input_data) async def on_input_data( msg: Data, logger: Logger, key: bytes Context(message.raw_message.key), ) - None: logger.info(on_input_data(msg%s), msg) await to_output_data.publish(Data(datamsg.data 1.0), keybkey) broker.subscriber(output_data) async def on_output_data( msg: Data, logger: Logger, key: bytes Context(message.raw_message.key), ) - None: logger.info(on_output_data(msg%s), msg)这里通过Context(message.raw_message.key)注入原始 Kafka 消息的 key并在publish(...)调用中传keybkey即可在消费侧同样用 Context 读取 key实现按 key 对齐的分区读写。发布路径的底层调用链综合源码KafkaBroker的发布路径可以归纳为入口KafkaBroker.publish(...)broker.py或DefaultPublisher.publish(...)usecase.py命令构造将消息、topic、key、partition、headers、correlation_id 等封装为KafkaPublishCommandcorrelation_id未指定时调用self.config.id_generator()生成中间件包装_basic_publishpub_base.py按逆序将 broker 中间件包装到 producer 的publish调用上底层发送由AsyncConfluentFastProducerImpl执行实际的confluent_kafka发送并按no_confirm决定是否等待 Kafka 确认。KafkaBroker还额外提供publish_batch(*messages, topic...)与request(message, topic, ...)两个发布族方法见 broker.py前者批量发送多条消息后者执行 Request-Reply 并等待响应消息。它们同样封装为KafkaPublishCommand并复用中间件与 producer 链路。AsyncAPI 文档化机制三种方式在 AsyncAPI 文档中的表现截然不同broker.publish(...)不进入文档broker.publisher(topic)创建的 Publisher 对象 / 装饰器当 Broker 被包装进FastStream对象后会导出到 AsyncAPI 文档。其实现位于 faststream/confluent/publisher/specification.pyKafkaPublisherSpecification会为每个 publisher 生成一个规范条目默认名称形如{topic}:Publisher例如output_data:Publisher包含address发布目标 topicoperation.message.payload依据 publisher 的schema未显式指定时从处理函数返回类型解析生成的 payload 定义bindings.kafkaKafka Channel Binding携带topic信息。这解释了为什么官方文档强调构建完整服务时应使用 Publisher 对象或装饰器以便让发布接口也纳入 AsyncAPI 契约。内存测试TestKafkaBroker上面三个示例都使用了TestKafkaBroker进行内存测试无需真实 Kafka 集群async with TestKafkaBroker(broker): await broker.publish(msg, topicinput_data) handle_data.mock.assert_called_once_with(dict(msg))TestKafkaBroker会替换底层连接将订阅处理函数替换为可断言的mock对象同时发布消息也会被记录因此你可以像断言订阅一样断言发布例如上文测试中的to_output_data.mock.assert_called_once_with(dict(Data(data1.2)))。这使得三种发布方式都具备开箱即用的可测试性。对应测试文件可见 tests/docs/confluent/raw_publish/test_raw_publish.py 与 tests/docs/confluent/publisher_object/test_publisher_object.py。三种发布方式的选择建议方式语法AsyncAPI 文档化适用场景broker.publish(...)await broker.publish(msg, topic...)否一次性消息、集成其它框架应用、临时发送Publisher 对象p broker.publisher(t); await p.publish(msg)是需要复用发布目标、在服务内部多处发布Publisher 装饰器broker.publisher(t)叠加在 subscriber 函数上是快速构建「订阅-处理-再发布」管道自动发布函数返回值在发布参数定制方面KafkaBroker构造函数统一配置acks、compression_type、partitioner、max_request_size、linger_ms、enable_idempotence等生产者参数单次发布时则通过key、partition、timestamp_ms、headers、correlation_id、reply_to、no_confirm精细控制。结合 官方发布基础文档 中关于序列化与correlation_id的约定即可在真实项目中落地稳定、可追踪、可文档化的 Kafka 发布能力。【免费下载链接】faststreamAsynchronous Python framework for event-driven services. A thin client for Kafka, RabbitMQ, NATS, Redis and MQTT with full access to native broker features, plus AsyncAPI docs, in-memory tests and observability out of the box.项目地址: https://gitcode.com/GitHub_Trending/fa/faststream创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考