FastStream Publisher Object 完全指南：可复用发布器、correlation

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 Object：

publisher = 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 / Confluent（Kafka 协议）

示例文件：kafka/object.py 与 confluent/object.py：

from 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.py：

from faststream import FastStream from faststream.rabbit import RabbitBroker broker = RabbitBroker("amqp://guest:guest@localhost: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.py：

from 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.py：

from 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.py：

from faststream import FastStream from faststream.mqtt import MQTTBroker broker = MQTTBroker("localhost", port=1883) 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_id：

@publisherproperly sets the samecorrelation_idas the incoming message.

这意味着当一条消息在多个服务间流转时，每个服务通过 Publisher Object 发出的下游消息都会携带与入站消息相同的correlation_id，从而在整个消息管道中形成一条可追踪的链路，便于日志聚合与链路追踪（trace）收集。

从源码看，这一机制在订阅器消费流程中实现：faststream/_internal/endpoint/subscriber/usecase.py 在处理函数返回后检查结果消息，若其correlation_id为空，则回填为入站消息的correlation_id：

if 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 Object

Publisher 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("", topic="test-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("", topic="test-topic", correlation_id="42") await publisher.assert_called_once_with("Hi!", correlation_id="42")

可用的断言能力

断言方式	作用
`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_id="42"，那么发布消息的断言也必须带上correlation_id="42"。

测试模式的底层机制

测试能力由 faststream/_internal/testing/calls.py 实现：

CallRecorder（calls.py#L158）负责记录端点看到的每条消息：record()中解码消息体并追加到calls列表、调用mock(decoded)记录调用；
CallAssertions（calls.py#L17）提供mock属性与assert_called_once_with/assert_called_with/assert_any_call三个断言方法；
发布器在测试模式下通过PublisherUsecase.set_test（publisher/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.py：PublisherUsecase._basic_publish通过_build_middlewares_stack把 broker 级发布中间件逐层包裹在 producer 调用之外，_basic_request则额外处理响应消息的解析与解码；
faststream/_internal/broker/pub_base.py：BrokerPublishMixin._basic_publish展示了 broker 层面的通用发布链路——从producer.publish出发，逆序叠加中间件后执行PublishCommand，同时提供publish_batch（批量发布，由具体 broker 决定是否支持）与request（RPC 请求）抽象；
faststream/_internal/endpoint/publisher/fake.py：FakePublisher是仅供 RPC / reply-to 应答使用的发布器实现，其publish/request方法会直接抛出NotImplementedError，提示只能在订阅器流程内用于响应消息，避免误用。

小结

Publisher Object 是 FastStream 推荐的“全功能”发布方式：一次定义、随处复用；作为装饰器时自动继承入站消息的correlation_id打通链路追踪；叠加多个发布器即可实现消息广播；配合 TestBroker 可在无外部 broker 的情况下完成消息体与字段级断言。上述示例与测试代码均可在 docs_src/getting_started/publishing 目录下找到，六种 broker（Kafka、Confluent、RabbitMQ、NATS、Redis、MQTT）的写法保持一致，可直接作为开发与测试的起点。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考