Невозможно показать сообщение в потребителе Kafka ⇐ Python

[Anonymous]

Я читаю API финансов Yahoo, чтобы получить несколько цен на акции и отправить их в kafkaProducer, который работает правильно.
код приведен ниже

Код: Выделить всё

 import yfinance as yf
from concurrent.futures import ThreadPoolExecutor
import threading
from confluent_kafka import Producer, Consumer

DELAY = 1.0  # Time between calls in seconds.

def get_stats(ticker):
info = yf.Tickers(ticker).tickers[ticker].info
print(f"{ticker} {info['currentPrice']}")
p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce(ticker, info['currentPrice'])
p.flush()

def call_API(event):
ticker_list = ['DLF.NS','SBIN.NS']

with ThreadPoolExecutor() as executor:
executor.map(get_stats, ticker_list)

event.set()  # Signal call has been made.

print('Running')
event = threading.Event()
while True:
event.clear()
timer = threading.Timer(DELAY, call_API, (event,))  # Call after delay.
timer.start()
event.wait()  # Blocks until event is set by another thread.

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

Код: Выделить всё

   from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': "stock"})
c.subscribe(['DLF.NS','SBIN.NS'])

while True:
msg = c.poll(1.0)

if msg is None:
continue
if msg.error():
print('Error: {}'.format(msg.error()))
continue

print('Received message on {}: {}'.format(msg.topic(), msg.value().decode('utf-8')))

Какие изменения мне следует внести, чтобы данные по всем темам были доступны пользователю

Подробнее здесь: https://stackoverflow.com/questions/783 ... a-consumer