2

我目前正在研究一个用例，使用 Kafka 和 robinhood 的浮士德来处理来自 Kafka 的数据。我已经成功地进行了计算，并且我需要的结果正在打印到我的浮士德工作者正在运行的控制台上。

现在我想找到一种方法让我的结果不仅在控制台中而且在 HTML 页面中可见。我已经查看了 websockets 库，但我无法让它与 faust 一起工作。我得到的错误是Crashed reason=RuntimeError('This event loop is already running')我认为这是因为代码是针对正在处理的每条消息执行的。

非常感谢任何帮助

这是我正在使用的代码：

    import faust, datetime, websockets, asyncio

app = faust.App(
    'UseCase',
    broker='kafka://localhost:29092',
)

usecase_topic = app.topic('usecase',partitions=8)

usecase_table = app.Table('usecase', default=int)

checkfailure = {}

@app.agent(usecase_topic)
async def process_record(records):
    async for record in records:
        #count records for each Sensor
        print(record)
        sensor = record['ext_id']
        usecase_table[sensor] += 1
        #print(f'Records for Sensor {sensor}: {usecase_table[sensor]}')

        #write current timestamp of record and previous timestamp for each sensor to usecase_table dict
        currtime_id = record['ext_id']+'c'
        prevtime_id = record['ext_id']+'p'
        usecase_table[currtime_id] = datetime.datetime.strptime(record['tag_tsp'], "%Y%m%d%H%M%S.%f")

        #print current time
        print(f'Current time for Sensor {sensor}: {usecase_table[currtime_id]}')


        #calculate and print timestamp delta; if no previous value is given print message
        if usecase_table[prevtime_id] == 0:
            print(f'no previous timestamp for sensor {sensor}')
        else:
            usecase_table[prevtime_id] = datetime.datetime.strptime(usecase_table[prevtime_id], "%Y%m%d%H%M%S.%f")
            print(f'previous time for Sensor {sensor}: {usecase_table[prevtime_id]}')
            tsdelta = usecase_table[currtime_id] - usecase_table[prevtime_id]
            tsdelta_id = record['ext_id']+'t'
            usecase_table[tsdelta_id] = str(tsdelta)
            print(f'Sensor: {sensor} timestamp delta: {usecase_table[tsdelta_id]}')

        #calculate value delta
        currvalue_id = record['ext_id']+'cv'
        prevvalue_id = record['ext_id']+'pv'
        usecase_table[currvalue_id] = record['tag_value_int']

        print(f'current value for Sensor {sensor}: {usecase_table[currvalue_id]}')

        if usecase_table[prevvalue_id] == 0:
            print(f'no previous record for sensor {sensor}')
        else:
            print(f'previous value for Sensor {sensor}: {usecase_table[prevvalue_id]}')
            vdelta = usecase_table[currvalue_id] - usecase_table[prevvalue_id]
            vdelta_id = record['ext_id']+'v'
            usecase_table[vdelta_id] = vdelta
            print(f'Sensor: {sensor} value delta:{usecase_table[vdelta_id]}')

        #calculate cycle time
        if usecase_table[prevtime_id] != 0 and usecase_table[prevvalue_id] != 0 and usecase_table[vdelta_id] != 0:
            cycletime = tsdelta / usecase_table[vdelta_id]
            cyclemsg = f'Sensor {sensor}; Cycletime {cycletime}'
            print(cyclemsg)

        #add timestamp to checkfailure dict
        checkfailure[sensor] = datetime.datetime.strptime(record['tag_tsp'], "%Y%m%d%H%M%S.%f")
        #check if newest timestamp for a sensor is older than 10 secs
        for key in checkfailure:
            if datetime.datetime.now() - checkfailure[key] >= datetime.timedelta(seconds=10):
                failuremsg = f'Error: Sensor {key}'
                print(failuremsg)

        #send results to websocket
        async def send_result(websocket,path):
            results = cyclemsg + failuremsg
            await websockets.send(results)
        start_server = websockets.serve(send_result, '127.0.0.1', 5678)
        asyncio.get_event_loop().run_until_complete(start_server)

        #set previous value and timestamp to current
        usecase_table[prevtime_id] = record['tag_tsp']
        usecase_table[prevvalue_id] = record['tag_value_int']
4

1 回答 1

9

被这个 asyncio 错误信息弄糊涂是正常的 :)

您不能loop.run_until_complete从async def函数调用。

您需要做的是在后台启动 websocket 服务器。这应该很容易，而且它正在使用asyncio.ensure_future，但是您还希望您的 websocket 服务器在您的应用程序退出时正常关闭。

出于这个原因，浮士德使用“服务”，你可以为你的 websocket 服务器定义一个服务：

import faust
import websockets
from mode import Service
from websockets.exceptions import ConnectionClosed
from websockets.server import WebSocketServerProtocol


class App(faust.App):

   def on_init(self):
       self.websockets = Websockets(self)

   async def on_start(self):
       await self.add_runtime_dependency(self.websockets)



class Websockets(Service):

    def __init__(self, app, bind: str = 'localhost', port: int = 9999, **kwargs):
        self.app = app
        self.bind = bind
        self.port = port
        super().__init__(**kwargs)

    async def on_message(self, ws, message):
        ...

    async def on_messages(self,
                          ws: WebSocketServerProtocol,
                          path: str) -> None:
        try:
            async for message in ws:
                await self.on_message(ws, message)
        except ConnectionClosed:
            await self.on_close(ws)
        except asyncio.CancelledError:
            pass

    async def on_close(self, ws):
        # called when websocket socket is closed.
        ...

    @Service.task
    def _background_server(self):
         await websockets.serve(self.on_messages, self.bind, self.port)

app = App('UseCase')
# [...]
于 2019-04-01T20:31:17.600 回答