Как создать несколько производителей и потребителей в Кафке с использованием Python - PullRequest
Новая
       8

Как создать несколько производителей и потребителей в Кафке с использованием Python

0 голосов
/ 28 февраля 2019

Я пытаюсь захватить поток с нескольких веб-камер, используя ссылку rtsp в приложении фляги, и хочу показать через браузер.Чтобы добиться этого, я создал двух отдельных производителей, тему и два потребителя.После запуска сервера kafka и запуска и customer.py, и provider.py оба потока работают только в течение двух секунд.

Я хочу захватить поток с нескольких веб-камер

Manufacturer.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()

и это мой потребитель.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)

1 Ответ

0 голосов
/ 21 августа 2019 
data = cv2.imencode('.jpeg', frame)[1].tobytes()

мой пример кода 

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

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


delete '[1]'
...