如何在 nameko 中异步进行 RPC 调用?

Question

我正在使用nameko构建一个具有微服务架构的ETL管道，我不想在发出RPC请求后等待回复。

from nameko.rpc import rpc, RpcProxy
class Scheduler(object):
name = "scheduler"
task_runner = RpcProxy('task_runner')
@rpc
def schedule(self, task_type, group_id, time):
return self.task_runner.start.async(task_type, group_id)

此代码引发错误：

Traceback (most recent call last):
File "/home/satnam-sandhu/.anaconda3/envs/etl/bin/nameko", line 8, in <module>
sys.exit(main())
File "/home/satnam-sandhu/.anaconda3/envs/etl/lib/python3.8/site-packages/nameko/cli/main.py", line 112, in main
args.main(args)
File "/home/satnam-sandhu/.anaconda3/envs/etl/lib/python3.8/site-packages/nameko/cli/commands.py", line 110, in main
main(args)
File "/home/satnam-sandhu/.anaconda3/envs/etl/lib/python3.8/site-packages/nameko/cli/run.py", line 181, in main
import_service(path)
File "/home/satnam-sandhu/.anaconda3/envs/etl/lib/python3.8/site-packages/nameko/cli/run.py", line 46, in import_service
__import__(module_name)
File "./scheduler/service.py", line 15
return self.task_runner.start.async(task_type, group_id)
^
SyntaxError: invalid syntax

我是微服务和Nameko的新手，我也使用RabbitMQ作为队列服务。

Answer 1

我遇到了同样的问题;您需要将async方法替换为call_async方法，并使用result()检索数据。

文档
GitHub 问题

Answer 2

使用call_async而不是异步或为了获得更好的结果使用事件

从nameko.events import EventDispatcher， event_handler

@event_handler("service_a", "event_emit_name")
def get_result(self, payload):
#do_something...

和其他服务

from nameko.events import EventDispatcher, event_handler
@event_handler("service_a", "event_emit_name")
def return_result(self, payload):
#get payload and work over there