FastStream Publisher Object 完全指南可复用发布器、correlation_id 链路追踪与消息广播【免费下载链接】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 中的Publisher Object发布器对象它是由broker.publisher(...)创建、可复用、可作装饰器、自带 AsyncAPI 表示与完整测试能力的消息发布方案。读完本文你将掌握如何在 Kafka、RabbitMQ、NATS、Redis、MQTT 与 Confluent 六种 broker 上通过 Publisher Object 发布消息理解其correlation_id自动传播与消息广播机制并能使用 TestBroker 对发布行为进行断言验证。Publisher Object 是什么在 FastStream 中除了 直接调用broker.publish()之外更完整的发布方式是使用Publisher Objectpublisher broker.publisher(another-topic)这条语句创建一个绑定到目标队列/主题another-topic的可复用发布器对象。它与直接发布相比具备四个核心优势AsyncAPI 支持Publisher Object 拥有 AsyncAPI 表示可以在生成的异步 API 文档中渲染出该发布通道测试支持该方法拥有完整的 Testing 支持可配合 TestBroker 进行内存级断言Context 集成可以借助 FastStream 内置的 Context依赖注入容器 访问 broker 或其它外部服务可复用一个 Publisher Object 可以在多处重复使用。同时它有一个需要留意的取舍消息将“总是”被发布——只要被装饰的函数执行并返回发布动作必然发生不存在条件跳过机制。从源码结构看broker.publisher(...)的注册逻辑位于 faststream/_internal/broker/registrator.py所有 broker 与 router 共享该注册器创建出的发布器被加入_publishers集合与__persistent_publishers持久化列表因此在 broker 生命周期内可被反复调用。六个 Broker 的 Publisher Object 用法Publisher Object 的 API 在六种 broker 上完全一致仅连接配置与队列命名不同。以下代码来自 docs_src/getting_started/publishing 目录均可在本地直接运行验证。AIOKafka / ConfluentKafka 协议示例文件kafka/object.py 与 confluent/object.pyfrom faststream import FastStream from faststream.kafka import KafkaBroker # Confluent 版改为 from faststream.confluent import KafkaBroker broker KafkaBroker(localhost:9092) app FastStream(broker) publisher broker.publisher(another-topic) publisher broker.subscriber(test-topic) async def handle() - str: return Hi! broker.subscriber(another-topic) async def handle_next(msg: str): assert msg Hi!RabbitMQ示例文件rabbit/object.pyfrom faststream import FastStream from faststream.rabbit import RabbitBroker broker RabbitBroker(amqp://guest:guestlocalhost:5672/) app FastStream(broker) publisher broker.publisher(another-queue) publisher broker.subscriber(test-queue) async def handle() - str: return Hi! broker.subscriber(another-queue) async def handle_next(msg: str): assert msg Hi!NATS示例文件nats/object.pyfrom faststream import FastStream from faststream.nats import NatsBroker broker NatsBroker(nats://localhost:4222) app FastStream(broker) publisher broker.publisher(another-subject) publisher broker.subscriber(test-subject) async def handle() - str: return Hi! broker.subscriber(another-subject) async def handle_next(msg: str): assert msg Hi!Redis示例文件redis/object.pyfrom faststream import FastStream from faststream.redis import RedisBroker broker RedisBroker(redis://localhost:6379) app FastStream(broker) publisher broker.publisher(another-channel) publisher broker.subscriber(test-channel) async def handle() - str: return Hi! broker.subscriber(another-channel) async def handle_next(msg: str): assert msg Hi!MQTT示例文件mqtt/object.pyfrom faststream import FastStream from faststream.mqtt import MQTTBroker broker MQTTBroker(localhost, port1883) app FastStream(broker) publisher broker.publisher(another-topic) publisher broker.subscriber(test-topic) async def handle() - str: return Hi! broker.subscriber(another-topic) async def handle_next(msg: str): assert msg Hi!装饰器用法与调用顺序Publisher Object 可以作为装饰器直接作用于订阅处理函数publisher broker.publisher(another-topic) publisher broker.subscriber(test-topic) async def handle() - str: return Hi!关于装饰器顺序官方文档给出了明确的规则publisher与broker.subscriber(...)的先后顺序无关紧要两种写法等价publisher只能作用于已经被broker.subscriber(...)装饰过的函数——它必须依附于某个订阅处理器否则没有消息源驱动发布。其底层实现印证了这一点在 faststream/_internal/endpoint/publisher/usecase.py 中PublisherUsecase.__call__先通过super().__call__(func)取得或包装出处理函数再把发布器自身追加到handler._publishers列表该列表定义于 faststream/_internal/endpoint/call_wrapper.py。当订阅器真正消费到消息并执行完处理函数后faststream/_internal/endpoint/subscriber/usecase.py 会遍历h.handler._publishers对每一个发布器调用p._publish(...)把处理结果发出去。返回类型注解的强制约定!!! note 重要约定 Publisher 装饰器使用处理函数返回值的类型注解来对返回值进行类型转换cast后再发送因此请务必准确标注返回类型。例如示例中的async def handle() - str其返回值Hi!会按str序列化后发送到another-topic。correlation_id跨服务链路追踪publisher装饰器会自动继承入站消息的correlation_idpublisherproperly sets the samecorrelation_idas the incoming message.这意味着当一条消息在多个服务间流转时每个服务通过 Publisher Object 发出的下游消息都会携带与入站消息相同的correlation_id从而在整个消息管道中形成一条可追踪的链路便于日志聚合与链路追踪trace收集。从源码看这一机制在订阅器消费流程中实现faststream/_internal/endpoint/subscriber/usecase.py 在处理函数返回后检查结果消息若其correlation_id为空则回填为入站消息的correlation_idif not result_msg.correlation_id: result_msg.correlation_id message.correlation_id随后的所有发布包括 RPC 应答与各 Publisher Object都会携带该correlation_id发出。Message Broadcasting一次处理多点广播Publisher 装饰器可以叠加使用多次实现消息广播——将同一个处理函数的返回值发送到多个目标队列publisher1 publisher2 broker.subscriber(in) async def handle(msg) - str: return Response执行效果是handle的返回值Response会被复制并发送到publisher1与publisher2各自绑定的所有输出主题。同样地源码路径清晰__call__中handler._publishers.append(self)会依次把每个发布器加入处理函数的发布器列表订阅器消费时按列表顺序逐个_publish。!!! note RPC 模式下的广播 如果该订阅器以RPC请求-响应模式消费消息那么除了向RPC 应答通道返回回复外还会同时向所有叠加的 Publisher 广播结果。这意味着 RPC 调用方与下游消费者会同时收到该结果。测试 Publisher ObjectPublisher Object 的测试能力在 publishing/test.md 中有完整说明核心是通过Test*Broker将 broker 切换到内存模式无需真实的外部 broker 即可运行测试非常适合 CI 或本地开发环境。以下测试示例来自 kafka/object_testing.py其它 broker 的写法完全一致import pytest from faststream.kafka import TestKafkaBroker from .object import broker, publisher pytest.mark.asyncio async def test_handle(): async with TestKafkaBroker(broker) as br: await br.publish(, topictest-topic) publisher.mock.assert_called_once_with(Hi!) pytest.mark.asyncio async def test_message_fields(): async with TestKafkaBroker(broker) as br: await br.publish(, topictest-topic, correlation_id42) await publisher.assert_called_once_with(Hi!, correlation_id42)可用的断言能力断言方式作用publisher.mock.assert_called_once_with(Hi!)校验发布器恰好被调用一次且消息体为Hi!await publisher.assert_called_once_with(body, correlation_id..., headers..., reply_to..., content_type..., path...)同步校验消息体与消息字段correlation_id、headers 等await publisher.assert_called_with(...)校验最近一次调用的消息await publisher.assert_any_call(...)校验调用历史中至少有一次匹配其中mock断言接收的可以是dict、Pydantic/msgspec 模型或 matcher字段断言则支持correlation_id、headers、reply_to、content_type、path、context等参数。这里publisher继承的是处理器所消费消息的correlation_id——例如test_message_fields中入站消息携带correlation_id42那么发布消息的断言也必须带上correlation_id42。测试模式的底层机制测试能力由 faststream/_internal/testing/calls.py 实现CallRecordercalls.py#L158负责记录端点看到的每条消息record()中解码消息体并追加到calls列表、调用mock(decoded)记录调用CallAssertionscalls.py#L17提供mock属性与assert_called_once_with/assert_called_with/assert_any_call三个断言方法发布器在测试模式下通过PublisherUsecase.set_testpublisher/usecase.py#L53-L65切换为测试状态。值得注意的细节Publisher 的 mock 并非仅记录publish方法的入参——测试 broker 会为输出主题建立一个虚拟消费者真实地消费发布出去的消息并存储该消费结果。因此断言针对的是“虚拟消费者实际收到的消息”而非“调用参数”。测试使用要点先创建后测试为了让发布器被测试 broker 正确 patch必须在运行测试 broker 之前创建好这些发布器即模块级创建publisher broker.publisher(...)端到端测试Test*Broker也支持配合真实外部 broker 使用使测试具备端到端能力详见 订阅器测试页面 中关于 Real Broker Testing 的说明Lifespan 中的发布如果发布器是由 lifespan 钩子触发而非订阅器触发则钩子必须在测试内部运行——使用TestApp即可参见 Events Testing。发布中间件与底层发布链路如需深入理解发布过程可以沿着以下源码链路继续阅读faststream/_internal/endpoint/publisher/usecase.pyPublisherUsecase._basic_publish通过_build_middlewares_stack把 broker 级发布中间件逐层包裹在 producer 调用之外_basic_request则额外处理响应消息的解析与解码faststream/_internal/broker/pub_base.pyBrokerPublishMixin._basic_publish展示了 broker 层面的通用发布链路——从producer.publish出发逆序叠加中间件后执行PublishCommand同时提供publish_batch批量发布由具体 broker 决定是否支持与requestRPC 请求抽象faststream/_internal/endpoint/publisher/fake.pyFakePublisher是仅供 RPC / reply-to 应答使用的发布器实现其publish/request方法会直接抛出NotImplementedError提示只能在订阅器流程内用于响应消息避免误用。小结Publisher Object 是 FastStream 推荐的“全功能”发布方式一次定义、随处复用作为装饰器时自动继承入站消息的correlation_id打通链路追踪叠加多个发布器即可实现消息广播配合 TestBroker 可在无外部 broker 的情况下完成消息体与字段级断言。上述示例与测试代码均可在 docs_src/getting_started/publishing 目录下找到六种 brokerKafka、Confluent、RabbitMQ、NATS、Redis、MQTT的写法保持一致可直接作为开发与测试的起点。【免费下载链接】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 Publisher Object 完全指南可复用发布器、correlation_id 链路追踪与消息广播【免费下载链接】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 中的Publisher Object发布器对象它是由broker.publisher(...)创建、可复用、可作装饰器、自带 AsyncAPI 表示与完整测试能力的消息发布方案。读完本文你将掌握如何在 Kafka、RabbitMQ、NATS、Redis、MQTT 与 Confluent 六种 broker 上通过 Publisher Object 发布消息理解其correlation_id自动传播与消息广播机制并能使用 TestBroker 对发布行为进行断言验证。Publisher Object 是什么在 FastStream 中除了 直接调用broker.publish()之外更完整的发布方式是使用Publisher Objectpublisher broker.publisher(another-topic)这条语句创建一个绑定到目标队列/主题another-topic的可复用发布器对象。它与直接发布相比具备四个核心优势AsyncAPI 支持Publisher Object 拥有 AsyncAPI 表示可以在生成的异步 API 文档中渲染出该发布通道测试支持该方法拥有完整的 Testing 支持可配合 TestBroker 进行内存级断言Context 集成可以借助 FastStream 内置的 Context依赖注入容器 访问 broker 或其它外部服务可复用一个 Publisher Object 可以在多处重复使用。同时它有一个需要留意的取舍消息将“总是”被发布——只要被装饰的函数执行并返回发布动作必然发生不存在条件跳过机制。从源码结构看broker.publisher(...)的注册逻辑位于 faststream/_internal/broker/registrator.py所有 broker 与 router 共享该注册器创建出的发布器被加入_publishers集合与__persistent_publishers持久化列表因此在 broker 生命周期内可被反复调用。六个 Broker 的 Publisher Object 用法Publisher Object 的 API 在六种 broker 上完全一致仅连接配置与队列命名不同。以下代码来自 docs_src/getting_started/publishing 目录均可在本地直接运行验证。AIOKafka / ConfluentKafka 协议示例文件kafka/object.py 与 confluent/object.pyfrom faststream import FastStream from faststream.kafka import KafkaBroker # Confluent 版改为 from faststream.confluent import KafkaBroker broker KafkaBroker(localhost:9092) app FastStream(broker) publisher broker.publisher(another-topic) publisher broker.subscriber(test-topic) async def handle() - str: return Hi! broker.subscriber(another-topic) async def handle_next(msg: str): assert msg Hi!RabbitMQ示例文件rabbit/object.pyfrom faststream import FastStream from faststream.rabbit import RabbitBroker broker RabbitBroker(amqp://guest:guestlocalhost:5672/) app FastStream(broker) publisher broker.publisher(another-queue) publisher broker.subscriber(test-queue) async def handle() - str: return Hi! broker.subscriber(another-queue) async def handle_next(msg: str): assert msg Hi!NATS示例文件nats/object.pyfrom faststream import FastStream from faststream.nats import NatsBroker broker NatsBroker(nats://localhost:4222) app FastStream(broker) publisher broker.publisher(another-subject) publisher broker.subscriber(test-subject) async def handle() - str: return Hi! broker.subscriber(another-subject) async def handle_next(msg: str): assert msg Hi!Redis示例文件redis/object.pyfrom faststream import FastStream from faststream.redis import RedisBroker broker RedisBroker(redis://localhost:6379) app FastStream(broker) publisher broker.publisher(another-channel) publisher broker.subscriber(test-channel) async def handle() - str: return Hi! broker.subscriber(another-channel) async def handle_next(msg: str): assert msg Hi!MQTT示例文件mqtt/object.pyfrom faststream import FastStream from faststream.mqtt import MQTTBroker broker MQTTBroker(localhost, port1883) app FastStream(broker) publisher broker.publisher(another-topic) publisher broker.subscriber(test-topic) async def handle() - str: return Hi! broker.subscriber(another-topic) async def handle_next(msg: str): assert msg Hi!装饰器用法与调用顺序Publisher Object 可以作为装饰器直接作用于订阅处理函数publisher broker.publisher(another-topic) publisher broker.subscriber(test-topic) async def handle() - str: return Hi!关于装饰器顺序官方文档给出了明确的规则publisher与broker.subscriber(...)的先后顺序无关紧要两种写法等价publisher只能作用于已经被broker.subscriber(...)装饰过的函数——它必须依附于某个订阅处理器否则没有消息源驱动发布。其底层实现印证了这一点在 faststream/_internal/endpoint/publisher/usecase.py 中PublisherUsecase.__call__先通过super().__call__(func)取得或包装出处理函数再把发布器自身追加到handler._publishers列表该列表定义于 faststream/_internal/endpoint/call_wrapper.py。当订阅器真正消费到消息并执行完处理函数后faststream/_internal/endpoint/subscriber/usecase.py 会遍历h.handler._publishers对每一个发布器调用p._publish(...)把处理结果发出去。返回类型注解的强制约定!!! note 重要约定 Publisher 装饰器使用处理函数返回值的类型注解来对返回值进行类型转换cast后再发送因此请务必准确标注返回类型。例如示例中的async def handle() - str其返回值Hi!会按str序列化后发送到another-topic。correlation_id跨服务链路追踪publisher装饰器会自动继承入站消息的correlation_idpublisherproperly sets the samecorrelation_idas the incoming message.这意味着当一条消息在多个服务间流转时每个服务通过 Publisher Object 发出的下游消息都会携带与入站消息相同的correlation_id从而在整个消息管道中形成一条可追踪的链路便于日志聚合与链路追踪trace收集。从源码看这一机制在订阅器消费流程中实现faststream/_internal/endpoint/subscriber/usecase.py 在处理函数返回后检查结果消息若其correlation_id为空则回填为入站消息的correlation_idif not result_msg.correlation_id: result_msg.correlation_id message.correlation_id随后的所有发布包括 RPC 应答与各 Publisher Object都会携带该correlation_id发出。Message Broadcasting一次处理多点广播Publisher 装饰器可以叠加使用多次实现消息广播——将同一个处理函数的返回值发送到多个目标队列publisher1 publisher2 broker.subscriber(in) async def handle(msg) - str: return Response执行效果是handle的返回值Response会被复制并发送到publisher1与publisher2各自绑定的所有输出主题。同样地源码路径清晰__call__中handler._publishers.append(self)会依次把每个发布器加入处理函数的发布器列表订阅器消费时按列表顺序逐个_publish。!!! note RPC 模式下的广播 如果该订阅器以RPC请求-响应模式消费消息那么除了向RPC 应答通道返回回复外还会同时向所有叠加的 Publisher 广播结果。这意味着 RPC 调用方与下游消费者会同时收到该结果。测试 Publisher ObjectPublisher Object 的测试能力在 publishing/test.md 中有完整说明核心是通过Test*Broker将 broker 切换到内存模式无需真实的外部 broker 即可运行测试非常适合 CI 或本地开发环境。以下测试示例来自 kafka/object_testing.py其它 broker 的写法完全一致import pytest from faststream.kafka import TestKafkaBroker from .object import broker, publisher pytest.mark.asyncio async def test_handle(): async with TestKafkaBroker(broker) as br: await br.publish(, topictest-topic) publisher.mock.assert_called_once_with(Hi!) pytest.mark.asyncio async def test_message_fields(): async with TestKafkaBroker(broker) as br: await br.publish(, topictest-topic, correlation_id42) await publisher.assert_called_once_with(Hi!, correlation_id42)可用的断言能力断言方式作用publisher.mock.assert_called_once_with(Hi!)校验发布器恰好被调用一次且消息体为Hi!await publisher.assert_called_once_with(body, correlation_id..., headers..., reply_to..., content_type..., path...)同步校验消息体与消息字段correlation_id、headers 等await publisher.assert_called_with(...)校验最近一次调用的消息await publisher.assert_any_call(...)校验调用历史中至少有一次匹配其中mock断言接收的可以是dict、Pydantic/msgspec 模型或 matcher字段断言则支持correlation_id、headers、reply_to、content_type、path、context等参数。这里publisher继承的是处理器所消费消息的correlation_id——例如test_message_fields中入站消息携带correlation_id42那么发布消息的断言也必须带上correlation_id42。测试模式的底层机制测试能力由 faststream/_internal/testing/calls.py 实现CallRecordercalls.py#L158负责记录端点看到的每条消息record()中解码消息体并追加到calls列表、调用mock(decoded)记录调用CallAssertionscalls.py#L17提供mock属性与assert_called_once_with/assert_called_with/assert_any_call三个断言方法发布器在测试模式下通过PublisherUsecase.set_testpublisher/usecase.py#L53-L65切换为测试状态。值得注意的细节Publisher 的 mock 并非仅记录publish方法的入参——测试 broker 会为输出主题建立一个虚拟消费者真实地消费发布出去的消息并存储该消费结果。因此断言针对的是“虚拟消费者实际收到的消息”而非“调用参数”。测试使用要点先创建后测试为了让发布器被测试 broker 正确 patch必须在运行测试 broker 之前创建好这些发布器即模块级创建publisher broker.publisher(...)端到端测试Test*Broker也支持配合真实外部 broker 使用使测试具备端到端能力详见 订阅器测试页面 中关于 Real Broker Testing 的说明Lifespan 中的发布如果发布器是由 lifespan 钩子触发而非订阅器触发则钩子必须在测试内部运行——使用TestApp即可参见 Events Testing。发布中间件与底层发布链路如需深入理解发布过程可以沿着以下源码链路继续阅读faststream/_internal/endpoint/publisher/usecase.pyPublisherUsecase._basic_publish通过_build_middlewares_stack把 broker 级发布中间件逐层包裹在 producer 调用之外_basic_request则额外处理响应消息的解析与解码faststream/_internal/broker/pub_base.pyBrokerPublishMixin._basic_publish展示了 broker 层面的通用发布链路——从producer.publish出发逆序叠加中间件后执行PublishCommand同时提供publish_batch批量发布由具体 broker 决定是否支持与requestRPC 请求抽象faststream/_internal/endpoint/publisher/fake.pyFakePublisher是仅供 RPC / reply-to 应答使用的发布器实现其publish/request方法会直接抛出NotImplementedError提示只能在订阅器流程内用于响应消息避免误用。小结Publisher Object 是 FastStream 推荐的“全功能”发布方式一次定义、随处复用作为装饰器时自动继承入站消息的correlation_id打通链路追踪叠加多个发布器即可实现消息广播配合 TestBroker 可在无外部 broker 的情况下完成消息体与字段级断言。上述示例与测试代码均可在 docs_src/getting_started/publishing 目录下找到六种 brokerKafka、Confluent、RabbitMQ、NATS、Redis、MQTT的写法保持一致可直接作为开发与测试的起点。【免费下载链接】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），仅供参考
