apache-flink - Flink 有状态函数，使用 python SDK 进行异步调用

Question

我正在使用 Python SDK 试用 Stateful Functions 2.1 API，但我看不到如何在不阻塞应用程序的情况下对外部 api 进行异步调用的明确方法。

这是可能的还是有人可以让我走上正确的道路？

Answer 1

从 StateFun 2.2.0 开始，您可以使用 AsyncRequestReplyHandler。

从文档：

Python SDK 附带了一个额外的处理程序 AsyncRequestReplyHandler，它支持 Python 的可等待函数（协程）。这个处理程序可以与异步 Python 框架一起使用，例如 aiohttp。

@functions.bind("example/hello")
async def hello(context, message):
   response = await compute_greeting(message)
   context.reply(response)

from aiohttp import web

handler = AsyncRequestReplyHandler(functions)

async def handle(request):
   req = await request.read()
   res = await handler(req)
   return web.Response(body=res, content_type="application/octet-stream")

app = web.Application()
app.add_routes([web.post('/statefun', handle)])

if __name__ == '__main__':
  web.run_app(app, port=5000)

你可以在这里看到一个完整的例子。