Celery任务中集成gRPC客户端遇InactiveRpcError问题求助

环境配置 #

启动Celery worker的命令：

celery -A worker.worker.celery_app worker --loglevel=info --pool=gevent --concurrency=500

问题现象 #

• gRPC客户端在Celery任务外可正常工作
• 不含gRPC客户端的Celery任务运行正常
• 在Celery任务中使用gRPC客户端时，抛出错误：

grpc._channel._InactiveRpcError: <grpc._channel._RPCState object at 0x7f0adb16d600>

相关代码 #

async def send_latest_grpc_packets(filter_type: PaquetTypeEnum, lastest_iec_packets: Optional[Iec104Packets] = None):
    with grpc.insecure_channel(f"{settings.GRPC_SERVER_HOST}:{settings.GRPC_SERVER_PORT}") as channel:
        stub = ioa_packets_pb2_grpc.PacketsStub(channel)
        if lastest_iec_packets is None:
            lastest_iec_packets = get_lastest_iec_packets(filter_type)
        formatted_iec_packets = format_iec_instance_to_schema(lastest_iec_packets)
        packets_request = ioa_packets_pb2.PacketsRequest(
            packets=cast_packet_to_proto_schema(formatted_iec_packets),
            type=filter_type,
        )
        response = stub.SendPackets(packets_request, timeout=30)

@celery_app.task()
def send_cyclical_TM_task():
    asyncio.run(send_latest_grpc_packets(PaquetTypeEnum.TM))

依赖版本 #

• grpcio = "^1.53.0"
• celery = "^5.2.7"

已尝试方案 #

将Celery执行池从prefork改为gevent，问题仍未解决。

临时解决方案 #

启动任务调用自身接口，通过该接口使用gRPC客户端，但希望找到更优方案，保持进程非阻塞且周期性执行。

求有gRPC与Celery集成经验的人帮忙分析原因并提供解决方案，谢谢！

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