Funboost RPC 模式

概述

RPC 模式让你获取已消费任务函数的返回值。发布者推送消息后得到 task_id，后续可查询执行结果。

核心原则： 在 BoosterParams 中设置 is_using_rpc_mode=True，然后用 AsyncResult（同步）或 AioAsyncResult（异步）获取结果。

适用场景

• 需要获取任务函数的返回值
• 实现请求-响应模式
• 构建等待后台任务完成的 API
• 编排依赖前置任务结果的工作流

前置条件

• 必须配置 Redis — RPC 结果存储在 Redis 中
• 必须设置 is_using_rpc_mode=True — 不设此项结果不会被持久化

速查表

方法	环境	返回
func.push(...)	同步	AsyncResult 对象
func.publish(msg, task_options=...)	同步	AsyncResult 对象
async_result.result	同步	阻塞直到结果就绪
async_result.status_and_result	同步	阻塞返回结果字典（超时返回 None）
await func.aio_push(...)	异步	AioAsyncResult 对象
await func.aio_publish(...)	异步	AioAsyncResult 对象
await aio_result.result	异步	等待直到结果就绪
await aio_result.status_and_result	异步	返回结果字典（超时返回 None）

同步 RPC 模式

前置条件： RPC 模式依赖 Redis 存储结果，需在 funboost_config.py 中配置 REDIS_HOST、REDIS_PORT、REDIS_PASSWORD。详见 funboost-broker-selection skill。

from funboost import boost, BoosterParams, BrokerEnum

@boost(BoosterParams(
    queue_name="add_queue",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    is_using_rpc_mode=True,  # ← 必须设置
))
def add(x: int, y: int):
    return x + y

if __name__ == "__main__":
    add.consume()

    async_result = add.push(3, 4)
    print(f"Task ID: {async_result.task_id}")
    print(f"结果: {async_result.result}")  # 阻塞等待 → 7

异步 RPC 模式

import asyncio
from funboost import boost, BoosterParams, BrokerEnum, ConcurrentModeEnum

@boost(BoosterParams(
    queue_name="async_add_queue",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    concurrent_mode=ConcurrentModeEnum.ASYNC,
    is_using_rpc_mode=True,
))
async def async_add(x: int, y: int):
    return x + y

async def main():
    async_add.consume()

    aio_result = await async_add.aio_push(10, 20)
    result = await aio_result.result  # 等待结果 → 30
    print(f"结果: {result}")

if __name__ == "__main__":
    asyncio.run(main())

根据 task_id 查询结果

from funboost import AsyncResult, AioAsyncResult

# 如果之前保存了 task_id
task_id = "some-task-id-string"

# 同步查询
result_obj = AsyncResult(task_id)
status_dict = result_obj.status_and_result  # 返回字典（超时返回 None）
print(status_dict['result'])    # 函数返回值
print(status_dict['success'])   # 是否成功

# 也可用 status_and_result_obj 获得对象（有更好的 IDE 补全）
status_obj = result_obj.status_and_result_obj  # 返回 FunctionResultStatus 对象
print(status_obj.result)
print(status_obj.success)

# 异步查询（须在 async def 内）
async def query_result():
    aio_obj = AioAsyncResult(task_id)
    status_dict = await aio_obj.status_and_result
    if status_dict:
        print(status_dict['result'])
    else:
        print("超时未获取到结果")

常见错误

错误	修正
忘记设置 is_using_rpc_mode=True	必须加到 BoosterParams 中——否则结果不会被存储
在 async 代码中用 AsyncResult.result	必须用 AioAsyncResult + await
没有配置 Redis	RPC 结果依赖 Redis，需配置 BrokerConnConfig.REDIS_*
期望 publish() 直接返回结果	publish()/push() 返回结果对象，需调用 .result
用 MEMORY_QUEUE 作为 broker 并期望 RPC	RPC 支持任何 broker，但结果始终存储在 Redis 中

超时行为： AsyncResult.result 默认等待 1800 秒（30 分钟），可通过 AsyncResult(task_id, timeout=30) 设置较短超时。

铁律

必须设置 is_using_rpc_mode=True。 不设置此项，框架不会持久化函数返回值。未设置时 status_and_result 超时返回 None，.result 将阻塞后抛出 HasNotAsyncResult 异常。

相关 Skill

• funboost-async-programming — async/await 异步编程
• funboost-faas-deploy — HTTP 微服务部署