测试gRPC asyncio双向流时请求未发送问题求助

环境信息 #

• Python版本：3.10
• 依赖包：
  • grpcio==1.49.1（基于HTTP/2的RPC框架）
  • grpcio-tools==1.49.1（gRPC的Protobuf代码生成工具）

问题描述 #

对接内部双向流RPC服务时，使用队列提供待发送请求，但请求从未通过网络传输，Wireshark监控1408端口无任何活动。运行代码后队列未被消费，最终抛出CancelledError异常。

示例代码 #

import asyncio
import logging
import queue
import threading
import time

# noinspection PyPackageRequirements
import grpc.aio

from coherence.services_pb2_grpc import NamedCacheServiceStub
from coherence.utils import JSONSerializer, RequestFactory

new_loop = asyncio.new_event_loop()
new_loop.set_debug(True)
sentinel = "poison"


def run_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()


async def run_queue(client_stub, send_queue):
    print("START LOOP")

    stream = client_stub.events(iter(send_queue.get, sentinel))  # initiate bidirectional streaming
    async for entry_response in stream:
        print(f"RESPONSE: {entry_response}")

    print("DONE LOOP")  # shouldn't be printed; queue should block if no items


def start_stream_handler(client_stub, send_queue):
    threading.Thread(target=lambda: run_loop(new_loop), daemon=True).start()
    asyncio.run_coroutine_threadsafe(run_queue(client_stub, send_queue), new_loop)


async def events_basic():
    channel = grpc.aio.insecure_channel("127.0.0.1:1408")
    send_queue = queue.Queue()

    cache_name = "foo"
    serializer = JSONSerializer()
    client_stub = NamedCacheServiceStub(channel)
    request_factory = RequestFactory(cache_name, "", serializer)

    start_stream_handler(client_stub, send_queue)
    time.sleep(5.0)

    print("Feeding the queue ...")
    send_queue.put_nowait(request_factory.map_event_subscribe())

    print("Giving time to drain request queue ...")

    time.sleep(5.0)
    print(f"Expected queue size is [0]; actual: [{send_queue.qsize()}]")
    print("EXIT")

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    asyncio.run(events_basic())

运行输出 #

DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:grpc._cython.cygrpc:Using AsyncIOEngine.POLLER as I/O engine
START LOOP
Feeding the queue ...
Giving time to drain request queue ...
Expected queue size is [0]; actual: [1]
EXIT
DEBUG:grpc.aio._call:Client request_iterator raised exception:
Traceback (most recent call last):
  File "/Users/rlubke/Library/Caches/pypoetry/virtualenvs/coherence-NZNGCFtv-py3.10/lib/python3.10/site-packages/grpc/aio/_call.py", line 417, in _consume_request_iterator
    await self._write(request)
  File "/Users/rlubke/Library/Caches/pypoetry/virtualenvs/coherence-NZNGCFtv-py3.10/lib/python3.10/site-packages/grpc/aio/_call.py", line 439, in _write
    await self._metadata_sent.wait()
  File "/Users/rlubke/.pyenv/versions/3.10.1/lib/python3.10/asyncio/locks.py", line 213, in wait
    await fut
asyncio.exceptions.CancelledError

原因分析 #

1. 跨事件循环的gRPC资源冲突：grpc.aio的通道和stub与创建它们的事件强绑定，代码中通道在主事件循环创建，却在另一个线程的事件循环中调用stub方法，导致gRPC内部异步操作无法正确关联，请求被取消。
2. 同步队列阻塞事件循环：使用queue.Queue（同步阻塞队列）作为请求迭代器，在异步上下文调用send_queue.get()会直接阻塞事件循环，导致gRPC无法处理请求发送的异步逻辑。
3. 程序退出未正确清理：主函数结束后主事件循环关闭，跨线程的gRPC调用被强制取消，触发CancelledError。

解决思路 #

1. 统一事件循环 #

所有gRPC相关操作（创建通道、stub、调用RPC）都在同一个事件循环中执行，避免跨线程混用。移除额外的线程和事件循环，直接在主循环中启动流处理协程。

2. 替换为异步队列 #

用asyncio.Queue替代queue.Queue，通过await queue.get()在异步上下文安全获取请求，避免阻塞事件循环。同时自定义异步生成器适配gRPC的请求迭代器要求：

async def request_generator(queue, sentinel):
    while True:
        item = await queue.get()
        if item == sentinel:
            break
        yield item

3. 规范程序退出逻辑 #

退出前发送哨兵值关闭请求生成器，等待流处理任务完成后再关闭通道，避免强制取消导致的异常。

修正后的示例代码 #

import asyncio
import logging

# noinspection PyPackageRequirements
import grpc.aio

from coherence.services_pb2_grpc import NamedCacheServiceStub
from coherence.utils import JSONSerializer, RequestFactory

sentinel = "poison"


async def request_generator(send_queue, sentinel):
    while True:
        item = await send_queue.get()
        if item == sentinel:
            break
        yield item


async def run_queue(client_stub, send_queue):
    print("START LOOP")
    try:
        stream = client_stub.events(request_generator(send_queue, sentinel))
        async for entry_response in stream:
            print(f"RESPONSE: {entry_response}")
    except asyncio.CancelledError:
        print("Stream cancelled")
    print("DONE LOOP")


async def events_basic():
    channel = grpc.aio.insecure_channel("127.0.0.1:1408")
    send_queue = asyncio.Queue()

    cache_name = "foo"
    serializer = JSONSerializer()
    client_stub = NamedCacheServiceStub(channel)
    request_factory = RequestFactory(cache_name, "", serializer)

    # 在主事件循环中启动协程
    stream_task = asyncio.create_task(run_queue(client_stub, send_queue))
    
    print("Feeding the queue ...")
    await send_queue.put(request_factory.map_event_subscribe())

    print("Giving time to drain request queue ...")
    await asyncio.sleep(5.0)
    
    print(f"Expected queue size is [0]; actual: [{send_queue.qsize()}]")
    
    # 发送哨兵值关闭流
    await send_queue.put(sentinel)
    await stream_task
    
    # 关闭通道
    await channel.close()
    print("EXIT")


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    asyncio.run(events_basic())

内容的提问来源于stack exchange，提问作者user1198243