3

我是 Python 新手，玩弄一些东西，尝试使用 Faust 通过 Kafka 与 Python 服务进行通信。所以我有一个小的 PoC 项目。浮士德应用定义：

# app.py

import faust as f
from models import ReadRequest, ReadResponse

app = f.App("faust-app", broker="kafka://localhost:9092", store="rocksdb://")

topics = {"read-request": app.topic("read-request", value_type=ReadRequest)}


def get_app() -> f.types.AppT:
    return app


def get_topic(name: str) -> f.types.TopicT:
    return topics[name]

我的数据库阅读器代理：

# reader.py

import pandas as pd
from pymongo import MongoClient
from app import get_app, get_topic

client = MongoClient()

app = get_app()
req_topic = get_topic("read-request")


@app.agent(req_topic)
async def read_request(requests):
    async for request in requests:
        db = client.test
        coll = db[request["collection"]]
        result = coll.find(request["query"])
        df = pd.DataFrame(result)
        response = {
            "id": request["id"],
            "data": list(df.loc[:, df.columns != "_id"].to_dict(orient="records")),
        }
        print(response) # debug <1>
        yield response

模型定义：

# models.py

import faust as f


class ReadRequest(f.Record):
    id: int
    collection: str
    query: dict

测试agent.ask()

# test.py

import asyncio
from reader import read_request
from models import ReadRequest


async def run():
    result = await read_request.ask(ReadRequest(id=1, collection="test", query={}))
    print(result) # debug <2>


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

所以我reader运行了 zookeeper、kafka 服务器、mongodb 和 faust worker。一切都在使用开箱即用的配置。

当我运行时，python3 test.py我看到debug <1>了预期的打印输出，但debug <2>永远不会关闭并且执行挂在那里。

浮士德医生说 在此处输入图像描述

所以我假设我做的一切都是正确的。有人有任何线索吗？

4

0 回答 0