Flask API 作为实时 kafka 消费者答案

【问题标题】：Flask API as real time kafka consumerFlask API 作为实时 kafka 消费者
【发布时间】：2017-04-17 02:27:30
【问题描述】：

我想构建一个使用 Flask 框架开发的 python API，它使用 Kafka 主题并将流推送到客户端（html 页面或其他应用程序）。

我尝试使用虚拟数据生成实时流程（请参阅下面的实时路线）。出现的问题是 result 变量仅在循环结束后推送，而 result 变量应在每次迭代时推送。

我还尝试使用 Kafka 连接生成实时流（请参阅下面的 kafka 路由）。问题是没有返回数据，而是请求没有完成。

from flask import Response, Flask
import time
from kafka import KafkaConsumer

application = Flask(__name__)

@application.route('/')
def index():
    return "Hello, World!"


@application.route('/realtime/')
def realtime():

    def createGenerator():

        for i in range(1,10):
            yield str(i) + '\n'
            time.sleep(0.2)

    return Response(createGenerator())


@application.route('/kafka/')
def kafkaStream():
    consumer = KafkaConsumer(bootstrap_servers = 'serverlocation',
                     client_id = 'name of client',
                     auto_offset_reset = 'earliest',
                     value_deserializer = lambda m: json.loads(m.decode('ascii')))

    consumer.subscribe(topics=['my-topic'])

    def events():
        result = []
        for message in consumer:
           if message is not None:
               result.append(message.value)
           yield result
    return Response(events())

if __name__ == '__main__':
    application.run(debug = True)

到目前为止，我有效地从 Kafka 接收数据的唯一方法是在控制台中打印结果。

from kafka import KafkaConsumer
consumer = KafkaConsumer(bootstrap_servers = 'serverlocation',
                     client_id = 'name of client',
                     auto_offset_reset = 'earliest',
                     value_deserializer = lambda m: json.loads(m.decode('ascii')))

consumer.subscribe(topics=['my-topic'])

for message in consumer:
    print message

我认为问题在于API无法推送数据，直到进程完成，并且因为KafkaConsumer连接是无限的，所以没有任何东西推送到客户端。

我该如何解决这个问题？

【问题讨论】：

对于其他人来说，答案是套接字

标签： python flask apache-kafka

【解决方案1】：

对于其他正在寻找解决此问题的方法的人。基于套接字的解决方案将在 kafka 消费者之上工作，该消费者要么不断地监听队列和帖子中的消息。

查看此链接了解更多信息。 kafka SocketIo

【讨论】：