如何使用python在kafka中创建多个生产者和消费者

Question

我正在尝试从多个摄像头捕捉流，使用烧瓶应用程序中的rtsp链接，并希望通过浏览器显示。为了实现这一点，我创建了两个单独的生产者，主题和两个消费者。在启动kafka服务器并同时运行consumer.py和producer.py之后，这两个流只运行2秒。

我想从多个摄像头中捕获流

producer.py

import time
import sys
import cv2
from kafka import KafkaProducer
from kafka.errors import KafkaError

producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer2 = KafkaProducer(bootstrap_servers='localhost:9092')
topic = 'my-topic'
topic2 = 'my-topic2'


def emit_video():
    print('start emitting')
    camera = cv2.VideoCapture('rtsp://webcam1')
    camera2 = cv2.VideoCapture('rtsp://webcam2')

    while True:
        success, frame = camera.read()
        success2, frame2 = camera2.read()
        if not success2:
            print("camera issue")
        # png might be too large to emit
        else:
            data = cv2.imencode('.jpeg', frame)[1].tobytes()
            data2 = cv2.imencode('.jpeg', frame2)[1].tobytes()

            future = producer.send(topic, data)
            future2 = producer2.send(topic2, data2)
            try:
                future.get(timeout=60)
                future2.get(timeout=60)
            except KafkaError as e:
                print(e)
                break

            print('.', end='', flush=True)
            # to reduce CPU usage
            # time.sleep(0.2)
        # print()
    # video.release()

    print('done')


emit_video()

，这是我的consumer.py

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

consumer = KafkaConsumer('my-topic', bootstrap_servers='localhost:9092')
consumer2 = KafkaConsumer('my-topic2', bootstrap_servers='localhost:9092')

app = Flask(__name__)


def kafkastream():
    for message in consumer:
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + message.value + b'\r\n\r\n')


def kafkastream2():
    print(threading.current_thread().getName())
    for message in consumer2:
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + message.value + b'\r\n\r\n')


@app.route('/video_feed')
def video_feed():
    return Response(kafkastream(),
                    mimetype='multipart/x-mixed-replace; boundary=frame')


@app.route('/video_feed2')
def video_feed2():
    return Response(kafkastream2(),
                    mimetype='multipart/x-mixed-replace; boundary=frame')


@app.route('/')
def index():
    return render_template('index.html')

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

Answer 1

data = cv2.imencode('.jpeg', frame)[1].tobytes()

我的示例代码

ret, buffer = cv2.imencode('.jpg', frame)

producer.send(topic, buffer.tobytes())


delete '[1]'