47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
import threading
|
|
from confluent_kafka import Consumer, KafkaException, KafkaError
|
|
|
|
def consume_messages():
|
|
# Настройка конфигурации потребителя
|
|
conf = {
|
|
'bootstrap.servers': 'localhost:9092',
|
|
'group.id': 'my_group',
|
|
'auto.offset.reset': 'earliest'
|
|
}
|
|
|
|
# Создание потребителя
|
|
consumer = Consumer(conf)
|
|
|
|
# Подписка на топик
|
|
consumer.subscribe(['my_topic'])
|
|
|
|
try:
|
|
while True:
|
|
# Получение сообщения
|
|
msg = consumer.poll(timeout=1.0)
|
|
if msg is None:
|
|
continue
|
|
if msg.error():
|
|
if msg.error().code() == KafkaError._PARTITION_EOF:
|
|
# Достигнут конец раздела
|
|
continue
|
|
else:
|
|
raise KafkaException(msg.error())
|
|
# Обработка сообщения
|
|
print(f'Received message: {msg.value().decode("utf-8")}')
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
# Закрытие потребителя
|
|
consumer.close()
|
|
|
|
def start_consumer_thread():
|
|
consumer_thread = threading.Thread(target=consume_messages)
|
|
consumer_thread.daemon = True
|
|
consumer_thread.start()
|
|
|
|
if __name__ == "__main__":
|
|
start_consumer_thread()
|
|
# Основной поток может выполнять другие задачи
|
|
while True:
|
|
pass |