Faust 传感器


class MySensor(Sensor):
    
    def on_message_in(self, tp: faust.types.tuples.TP, offset: int, message: faust.types.tuples.Message) -> None:
        print('=======================')
        print(message)

    def on_stream_event_in(self, tp: faust.types.tuples.TP, offset: int, stream: faust.types.streams.StreamT, event: faust.types.events.EventT) -> None:
        print('#################')
        print(event)


app = faust.App(
    'ht_app',
    broker=settings.KAFKA_URL
)

app.sensors.add(MySensor())