尧图网页设计 
FastStream Redis List 批量订阅与批量发布实战指南基于 ListSub 的高吞吐消息处理【免费下载链接】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/faststreamRedis List 是 FastStream 支持批量消费与批量发布的唯一数据结构。本文以ListSub(batchTrue)为核心完整讲解如何在 FastStream 中从 Redis List 批量拉取消息、用publish_batch一次性写入多条消息并结合仓库源码剖析底层LPOP/RPUSH的调用链与max_records、polling_interval等关键参数的实现细节帮助你构建高吞吐的 Redis 数据管道。为什么需要批量订阅 Redis ListRedis List 本质上是按插入顺序排列的字符串链表常被用作轻量级任务队列。逐条消费LPOP/BLPOP在消息量巨大时会产生大量网络往返与 Python 级调度开销而批量消费一次从队首取出多条消息交给处理函数能显著提升资源利用率与吞吐量。FastStream 在#!python broker.subscriber(...)装饰器中通过ListSub对象开启这一能力其完整用法正是本文要讲解的内容。核心概念ListSub订阅对象在 FastStream 中Redis List 的订阅参数由ListSub对象表达它定义于 faststream/redis/schemas/list_sub.py构造参数如下均为源码确认的默认值参数类型默认值作用list_namestr必填要订阅的 Redis List 名称batchboolFalse是否以批量模式消费True时处理器收到消息列表max_recordsint10每次批量拉取的最大消息条数仅在batchTrue时生效polling_intervalfloat0.1列表为空时轮询间隔秒批量模式下用于控制无数据时的休眠时长从源码看ListSub还暴露了records只读属性batchTrue时返回max_records否则返回None这直接影响底层每次拉取的数量# faststream/redis/schemas/list_sub.py property def records(self) - int | None: return self.max_records if self.batch else None这意味着默认每次批量最多取 10 条若你的处理逻辑需要更大批次请显式调大max_records。使用批量订阅消费 Redis List第一步定义带批量能力的订阅者在#!python broker.subscriber(...)中传入ListSub对象并将batch置为Truebroker.subscriber(listListSub(test-list, batchTrue)) async def handle(msg: list[str], logger: Logger): logger.info(msg)list参数指定订阅 Redis List而非 Pub/Sub 频道或 StreamListSub(test-list, batchTrue)明确告知框架消费test-list时以批量模式工作。第二步实现消费函数消费函数签名中的msg声明为list[str]也可结合 Pydantic 模型组成list[Model]装饰器会自动把一次批量拉取的多条消息聚合成列表后调用broker.subscriber(listListSub(test-list, batchTrue)) async def handle(msg: list[str], logger: Logger): logger.info(msg)完整可运行示例完整示例位于 docs/docs_src/redis/list/sub_batch.pyfrom faststream import FastStream, Logger from faststream.redis import ListSub, RedisBroker broker RedisBroker() app FastStream(broker) broker.subscriber(listListSub(test-list, batchTrue)) async def handle(msg: list[str], logger: Logger): logger.info(msg)RedisBroker()默认连接redis://localhost:6379如需指定地址可仿照 docs/docs_src/redis/list/list_pub.py 传入RedisBroker(redis://localhost:6379)。启动应用后向test-list写入的消息会被成批送到handle由logger.info(msg)输出整批内容。源码级原理解析批量消费的底层链路批量拉取LPOPcount批量订阅者的实现在 faststream/redis/subscriber/usecases/list_subscriber.py 的ListBatchSubscriber类中。其核心_get_msgs使用 Redis 的LPOP name count一次弹出最多max_records条async def _get_msgs(self, client: Redis[bytes]) - None: async with self._read_lock: raw_msgs await client.lpop( nameself.list_sub.name, countself.list_sub.max_records, ) if raw_msgs: msg BatchListMessage( typeblist, channelself.list_sub.name, dataraw_msgs, ) await self.consume_one(msg) if not raw_msgs: await anyio.sleep(self.list_sub.polling_interval)两个关键细节值得注意当列表为空时lpop返回None框架会按polling_interval默认 0.1 秒休眠后重试避免忙轮询空耗 CPU单条消费模式ListSubscriber则走BLPOP阻塞语义client.blpop(name, timeoutpolling_interval)超时后自动返回None进入下一轮。从源码结构看FastStream 会根据ListSub.batch的值在 faststream/redis/subscriber/factory.py 中选择实例化ListSubscriber还是ListBatchSubscriber二者共享_ListHandlerMixin提供的读取锁、优雅停机graceful_timeout等能力。批量解析逐条解码并携带 batch_headers批量消息的解析由 faststream/redis/parser/parsers.py 中的RedisBatchListParser完成。它遍历data中的每条原始字节逐条还原消息体与消息头把多个 body 打包为 JSON 数组并将每条消息独立的 headers存为batch_headers列表同时取第一条消息的 headers 作为整体消息的headersfor x in message[data]: msg_data, msg_headers _decode_batch_body_item(x, self.config.message_format) body.append(msg_data) batch_headers.append(msg_headers) first_msg_headers next(iter(batch_headers), {})这一设计意味着消费函数拿到的虽然是一个列表但通过RedisMessage消息对象仍可访问msg.batch_headers获取每条消息各自的元信息如correlation_id这在多消息关联追踪场景非常有用——测试 tests/brokers/redis/test_consume.py 的test_consume_list_batch_headers即验证了msg.headers[correlation_id] msg.batch_headers[0][correlation_id]的行为。批量发布一次 RPUSH 多条消息FastStream 官方文档明确指出Redis List 是唯一支持批量发布的数据结构。实现方式有两种。方式一直接调用broker.publish_batchawait broker.publish_batch(msg1, msg2, listtest-list)方法签名定义于 faststream/redis/broker/broker.py支持*messages变长消息体、list目标列表名、correlation_id、reply_to、headers以及可选的pipelineRedis Pipeline用于事务化批量写入。其返回值int为本次批量发布操作的结果即RPUSH后列表长度。底层实现在 faststream/redis/publisher/producer.py先把每个消息体经cmd.message_format.encode(...)序列化再一次性执行connection.rpush(cmd.destination, *batch)——所有消息在一次网络请求内完成入队async def publish_batch(self, cmd: RedisPublishCommand) - int: batch [ await cmd.message_format.encode( messagemsg, correlation_idcmd.correlation_id or , reply_tocmd.reply_to, headerscmd.headers, serializerself.serializer, codecself.codec, ) for msg in cmd.batch_bodies ] connection cmd.pipeline or self._connection.client return cast(int, await connection.rpush(cmd.destination, *batch))当传入pipeline时发布会追加到该 Pipeline 中延迟统一执行相关测试见 tests/brokers/redis/test_publish.py 的test_publish_batch_with_pipeline通过publish_batch(*range(5), list..., pipelinepipe)一次压入 5 条并与max_records5的批量订阅者配对验证。方式二声明一个批量发布器除了直接调用还可以像订阅一样用ListSub声明发布器publisher broker.publisher(listListSub(test-list, batchTrue)) # 之后在处理器中复用 await publisher.publish(msg1, msg2)这种写法让“批量”成为发布器的固有语义配合broker.publisher(...)装饰器可轻松构建“消费一批 → 处理 → 发布一批”的管道且publisher对象与publish_batch走同一套RPUSH底层实现。批量消费/发布与消息头的配合批量模式下消息头遵循以下约定由RedisBatchListParser实现决定处理器整体消息的headers取批内第一条消息的 headers批内每条消息各自的 headers 保存在msg.batch_headers列表顺序与 body 一致correlation_id在批量发布时若未显式指定会为整批生成并注入每条消息。若你的下游逻辑依赖逐条消息的追踪元数据应通过batch_headers按索引读取而不是依赖整体headers。测试验证与实战建议仓库测试 tests/brokers/redis/test_consume.py 的test_consume_list_batch_with_one展示了一个关键事实即使队列中只有 1 条消息批量订阅者也会立即触发处理器并收到[hi]这样的单元素列表——批量模式不会等待攒满max_records才回调而是“有多少取多少”最多max_records条。这给了我们三条实战建议按处理耗时调大max_records默认 10 条适合大部分场景若单条处理快而网络开销占比高可适当提高到 50100以摊薄每次回调的固定成本善用polling_interval控制空转低延迟场景调小如 0.01 秒见测试用例高吞吐场景保持默认 0.1 秒避免空轮询结合 Redis Pipeline 批量发布在需要原子性批量入队的场景将publish_batch挂到 Pipeline 上一次 RTT 完成整批写入。小结本文围绕 docs/docs/en/redis/list/batch.md 的核心内容展开通过ListSub(test-list, batchTrue)开启 Redis List 批量订阅消费函数以list[...]接收消息通过broker.publish_batch(...)或broker.publisher(listListSub(..., batchTrue))实现批量发布。结合源码可见其本质是LPOP name count批量拉取 RPUSH ...批量写入的封装max_records、polling_interval、batch_headers分别控制批次大小、空转轮询与逐条元信息。理解这一机制后你便能用最少的网络开销搭建出高效的 Redis List 数据处理管道。【免费下载链接】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 Redis List 批量订阅与批量发布实战指南：基于 ListSub 的高吞吐消息处理
FastStream Redis List 批量订阅与批量发布实战指南基于 ListSub 的高吞吐消息处理【免费下载链接】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/faststreamRedis List 是 FastStream 支持批量消费与批量发布的唯一数据结构。本文以ListSub(batchTrue)为核心完整讲解如何在 FastStream 中从 Redis List 批量拉取消息、用publish_batch一次性写入多条消息并结合仓库源码剖析底层LPOP/RPUSH的调用链与max_records、polling_interval等关键参数的实现细节帮助你构建高吞吐的 Redis 数据管道。为什么需要批量订阅 Redis ListRedis List 本质上是按插入顺序排列的字符串链表常被用作轻量级任务队列。逐条消费LPOP/BLPOP在消息量巨大时会产生大量网络往返与 Python 级调度开销而批量消费一次从队首取出多条消息交给处理函数能显著提升资源利用率与吞吐量。FastStream 在#!python broker.subscriber(...)装饰器中通过ListSub对象开启这一能力其完整用法正是本文要讲解的内容。核心概念ListSub订阅对象在 FastStream 中Redis List 的订阅参数由ListSub对象表达它定义于 faststream/redis/schemas/list_sub.py构造参数如下均为源码确认的默认值参数类型默认值作用list_namestr必填要订阅的 Redis List 名称batchboolFalse是否以批量模式消费True时处理器收到消息列表max_recordsint10每次批量拉取的最大消息条数仅在batchTrue时生效polling_intervalfloat0.1列表为空时轮询间隔秒批量模式下用于控制无数据时的休眠时长从源码看ListSub还暴露了records只读属性batchTrue时返回max_records否则返回None这直接影响底层每次拉取的数量# faststream/redis/schemas/list_sub.py property def records(self) - int | None: return self.max_records if self.batch else None这意味着默认每次批量最多取 10 条若你的处理逻辑需要更大批次请显式调大max_records。使用批量订阅消费 Redis List第一步定义带批量能力的订阅者在#!python broker.subscriber(...)中传入ListSub对象并将batch置为Truebroker.subscriber(listListSub(test-list, batchTrue)) async def handle(msg: list[str], logger: Logger): logger.info(msg)list参数指定订阅 Redis List而非 Pub/Sub 频道或 StreamListSub(test-list, batchTrue)明确告知框架消费test-list时以批量模式工作。第二步实现消费函数消费函数签名中的msg声明为list[str]也可结合 Pydantic 模型组成list[Model]装饰器会自动把一次批量拉取的多条消息聚合成列表后调用broker.subscriber(listListSub(test-list, batchTrue)) async def handle(msg: list[str], logger: Logger): logger.info(msg)完整可运行示例完整示例位于 docs/docs_src/redis/list/sub_batch.pyfrom faststream import FastStream, Logger from faststream.redis import ListSub, RedisBroker broker RedisBroker() app FastStream(broker) broker.subscriber(listListSub(test-list, batchTrue)) async def handle(msg: list[str], logger: Logger): logger.info(msg)RedisBroker()默认连接redis://localhost:6379如需指定地址可仿照 docs/docs_src/redis/list/list_pub.py 传入RedisBroker(redis://localhost:6379)。启动应用后向test-list写入的消息会被成批送到handle由logger.info(msg)输出整批内容。源码级原理解析批量消费的底层链路批量拉取LPOPcount批量订阅者的实现在 faststream/redis/subscriber/usecases/list_subscriber.py 的ListBatchSubscriber类中。其核心_get_msgs使用 Redis 的LPOP name count一次弹出最多max_records条async def _get_msgs(self, client: Redis[bytes]) - None: async with self._read_lock: raw_msgs await client.lpop( nameself.list_sub.name, countself.list_sub.max_records, ) if raw_msgs: msg BatchListMessage( typeblist, channelself.list_sub.name, dataraw_msgs, ) await self.consume_one(msg) if not raw_msgs: await anyio.sleep(self.list_sub.polling_interval)两个关键细节值得注意当列表为空时lpop返回None框架会按polling_interval默认 0.1 秒休眠后重试避免忙轮询空耗 CPU单条消费模式ListSubscriber则走BLPOP阻塞语义client.blpop(name, timeoutpolling_interval)超时后自动返回None进入下一轮。从源码结构看FastStream 会根据ListSub.batch的值在 faststream/redis/subscriber/factory.py 中选择实例化ListSubscriber还是ListBatchSubscriber二者共享_ListHandlerMixin提供的读取锁、优雅停机graceful_timeout等能力。批量解析逐条解码并携带 batch_headers批量消息的解析由 faststream/redis/parser/parsers.py 中的RedisBatchListParser完成。它遍历data中的每条原始字节逐条还原消息体与消息头把多个 body 打包为 JSON 数组并将每条消息独立的 headers存为batch_headers列表同时取第一条消息的 headers 作为整体消息的headersfor x in message[data]: msg_data, msg_headers _decode_batch_body_item(x, self.config.message_format) body.append(msg_data) batch_headers.append(msg_headers) first_msg_headers next(iter(batch_headers), {})这一设计意味着消费函数拿到的虽然是一个列表但通过RedisMessage消息对象仍可访问msg.batch_headers获取每条消息各自的元信息如correlation_id这在多消息关联追踪场景非常有用——测试 tests/brokers/redis/test_consume.py 的test_consume_list_batch_headers即验证了msg.headers[correlation_id] msg.batch_headers[0][correlation_id]的行为。批量发布一次 RPUSH 多条消息FastStream 官方文档明确指出Redis List 是唯一支持批量发布的数据结构。实现方式有两种。方式一直接调用broker.publish_batchawait broker.publish_batch(msg1, msg2, listtest-list)方法签名定义于 faststream/redis/broker/broker.py支持*messages变长消息体、list目标列表名、correlation_id、reply_to、headers以及可选的pipelineRedis Pipeline用于事务化批量写入。其返回值int为本次批量发布操作的结果即RPUSH后列表长度。底层实现在 faststream/redis/publisher/producer.py先把每个消息体经cmd.message_format.encode(...)序列化再一次性执行connection.rpush(cmd.destination, *batch)——所有消息在一次网络请求内完成入队async def publish_batch(self, cmd: RedisPublishCommand) - int: batch [ await cmd.message_format.encode( messagemsg, correlation_idcmd.correlation_id or , reply_tocmd.reply_to, headerscmd.headers, serializerself.serializer, codecself.codec, ) for msg in cmd.batch_bodies ] connection cmd.pipeline or self._connection.client return cast(int, await connection.rpush(cmd.destination, *batch))当传入pipeline时发布会追加到该 Pipeline 中延迟统一执行相关测试见 tests/brokers/redis/test_publish.py 的test_publish_batch_with_pipeline通过publish_batch(*range(5), list..., pipelinepipe)一次压入 5 条并与max_records5的批量订阅者配对验证。方式二声明一个批量发布器除了直接调用还可以像订阅一样用ListSub声明发布器publisher broker.publisher(listListSub(test-list, batchTrue)) # 之后在处理器中复用 await publisher.publish(msg1, msg2)这种写法让“批量”成为发布器的固有语义配合broker.publisher(...)装饰器可轻松构建“消费一批 → 处理 → 发布一批”的管道且publisher对象与publish_batch走同一套RPUSH底层实现。批量消费/发布与消息头的配合批量模式下消息头遵循以下约定由RedisBatchListParser实现决定处理器整体消息的headers取批内第一条消息的 headers批内每条消息各自的 headers 保存在msg.batch_headers列表顺序与 body 一致correlation_id在批量发布时若未显式指定会为整批生成并注入每条消息。若你的下游逻辑依赖逐条消息的追踪元数据应通过batch_headers按索引读取而不是依赖整体headers。测试验证与实战建议仓库测试 tests/brokers/redis/test_consume.py 的test_consume_list_batch_with_one展示了一个关键事实即使队列中只有 1 条消息批量订阅者也会立即触发处理器并收到[hi]这样的单元素列表——批量模式不会等待攒满max_records才回调而是“有多少取多少”最多max_records条。这给了我们三条实战建议按处理耗时调大max_records默认 10 条适合大部分场景若单条处理快而网络开销占比高可适当提高到 50100以摊薄每次回调的固定成本善用polling_interval控制空转低延迟场景调小如 0.01 秒见测试用例高吞吐场景保持默认 0.1 秒避免空轮询结合 Redis Pipeline 批量发布在需要原子性批量入队的场景将publish_batch挂到 Pipeline 上一次 RTT 完成整批写入。小结本文围绕 docs/docs/en/redis/list/batch.md 的核心内容展开通过ListSub(test-list, batchTrue)开启 Redis List 批量订阅消费函数以list[...]接收消息通过broker.publish_batch(...)或broker.publisher(listListSub(..., batchTrue))实现批量发布。结合源码可见其本质是LPOP name count批量拉取 RPUSH ...批量写入的封装max_records、polling_interval、batch_headers分别控制批次大小、空转轮询与逐条元信息。理解这一机制后你便能用最少的网络开销搭建出高效的 Redis List 数据处理管道。【免费下载链接】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），仅供参考