forked from Alexey/DAS_2023_1
34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
|
from celery import Celery
|
|||
|
|
import time
|
|||
|
|
import pika
|
|||
|
|
|
|||
|
|
app = Celery('consumer', broker='pyamqp://guest:guest@localhost//')
|
|||
|
|
|
|||
|
|
|
|||
|
|
@app.task
|
|||
|
|
def process_messages(queue_name, delay):
|
|||
|
|
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
|
|||
|
|
channel = connection.channel()
|
|||
|
|
|
|||
|
|
# Объявляем очередь с именем
|
|||
|
|
channel.queue_declare(queue=queue_name)
|
|||
|
|
|
|||
|
|
# Привязываем очередь к 'logs' exchange
|
|||
|
|
channel.queue_bind(exchange='logs', queue=queue_name)
|
|||
|
|
|
|||
|
|
def callback(ch, method, properties, body):
|
|||
|
|
print(f" [x] Получено сообщение: {body}")
|
|||
|
|
time.sleep(delay)
|
|||
|
|
print(f" [x] Обработано сообщение: {body}")
|
|||
|
|
|
|||
|
|
# Устанавливаем consumer а чтобы получать сообщения из очереди
|
|||
|
|
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
|
|||
|
|
|
|||
|
|
print(f' [*] Ожидаем сообщения от {queue_name}')
|
|||
|
|
channel.start_consuming()
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == '__main__':
|
|||
|
|
process_messages.delay('consumer1_queue', 2) # Queue для Consumer 1 с задержкой 2 с
|
|||
|
|
process_messages.delay('consumer2_queue', 0) # Queue для Consumer 2 без задержки
|