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-selectionskill。
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 微服务部署