Merge pull request 'balakhonov_danila_lab_4' (#374) from balakhonov_danila_lab_4 into main

Reviewed-on: #374

balakhonov_danila_lab_4/.gitignore

@@ -0,0 +1 @@
+__pycache__

balakhonov_danila_lab_4/README.md

@@ -0,0 +1,86 @@
+# Лабораторная работа номер 4
+
+> Здравствуйте меня зовут Балахонов Данила группа ПИбд-42
+> 
+> *— Балахонов Данила ПИбд-42*
+
+Видео лабораторной работы номер 4 доступно по этой [ссылке](https://drive.google.com/file/d/1OZJ67OmOnQCAAb2VVg4OV7qMlZ1Y_HsH/view?usp=sharing).
+
+## Как запустить лабораторную работу номер 4?
+### Необходимые компоненты для запуска лабораторной работы номер 4
+> Здесь рассказана установка необходимых компонентов для запуска лабораторной работы номер 3 под дистрибутив GNU/Linux **Ubuntu**.
+
+Для запуска лабораторной работы номер 4 необходимы такие компоненты: 
+ - Git
+ - Docker
+ - Docker compose
+
+Чтобы установить **Git**, необходимо ввести данные команды в командную строку:
+``` bash
+sudo apt-get update
+sudo apt-get install git
+```
+
+Чтобы установить **Docker** и **Docker compose**, стоит ввести такие команды:
+``` bash
+# Настройка репозитория Docker
+sudo apt-get update
+sudo apt-get install ca-certificates curl
+sudo install -m 0755 -d /etc/apt/keyrings
+sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
+sudo chmod a+r /etc/apt/keyrings/docker.asc
+
+echo \
+  "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/ubuntu \
+  $(. /etc/os-release && echo "$VERSION_CODENAME") stable" | \
+  sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
+sudo apt-get update
+
+# Установка Docker и его компонентов
+sudo apt-get install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
+```
+
+Помимо Docker необходимо установить `python3`.
+``` bash
+sudo apt install python3
+```
+### Запуск лабораторной работы номер 4
+Для запуска лабораторной работы номер 4 необходимо **склонировать** репозиторий в любую папку и **перейти на ветку** balakhonov_danila_lab_4.
+Далее в папке с `docker-compose.yaml` нужно вызвать такую команду:
+``` bash
+sudo docker compose up -d
+```
+Таким образом контейнер с редисом будет запущен в фоне.
+
+Чтобы запустить программу, нужно вызывать три разных файла:
+- Файл *издателя*
+``` bash
+python3 program/publisher.py
+```
+- Файл *первого потребителя* с задержкой
+``` bash
+python3 program/consumer_1.py
+```
+- Файл *второго потребителя*
+``` bash
+python3 program/consumer_2.py
+```
+## Какие технологии были использованы?
+Для выполнения лабораторной работы номер 4 были использованы такие технологии, как:
+ - Docker compose
+ - Git
+ - Python 3
+ - Pika
+ - RabbitMQ
+## Что делает лабораторная работа номер 4?
+Суть лабораторной работы номер 4 заключается в получении навыков работы с брокером сообщений.
+## Вывод лабораторной работы номер 4
+Были пройдены туториалы по ознакомлению с RabbitMQ:
+### 1 туториал
+![1 туториал](imgs/tutor1.png)
+### 2 туториал
+![2 туториал](imgs/tutor2.png)
+### 3 туториал
+![3 туториал](imgs/tutor3.png)
+
+Также была разработана программа, которая состоит из издателя и двух потребителей. Разница между двумя потребителями заключается в скорости их обработки. Первый потребитель имеет некоторую задержку, поэтому он обрабатывает полученные данные гораздо дольше, чем второй потребитель. Из-за задержки происходит накапливание сообщений в очереди первого потребителя. Так как второй потребитель не имеет задержек, то и его очередь не будет настолько заметно возрастать. Если запустить несколько первых потребителей, то очередь также накапливается, но она меньше, так как уже два потребителя принимают сообщения.

balakhonov_danila_lab_4/docker-compose.yaml

@@ -0,0 +1,10 @@
+services:
+  rabbitmq:
+    image: rabbitmq:3-management-alpine
+    container_name: 'rabbitmq'
+    ports:
+        - "5672:5672"
+        - "15672:15672"
+    volumes:
+        - ~/.docker-conf/rabbitmq/data/:/var/lib/rabbitmq/
+        - ~/.docker-conf/rabbitmq/log/:/var/log/rabbitmq

balakhonov_danila_lab_4/program/consumer_1.py

@@ -0,0 +1,28 @@
+import pika
+import time
+from publisher import EXCHANGE
+
+QUEUE = 'consumer_1_queue'
+
+def callback_1(ch, method, properties, body):
+    print(f'[x] CONSUMER 1: GET. {body.decode()}')
+    # Ожидаем некоторое время
+    time.sleep(3)
+    print('[!] CONSUMER 1 OFF')
+
+
+def consume(callback, queue, exchange):
+    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
+    channel = connection.channel()
+
+    channel.queue_declare(queue=queue)
+    channel.queue_bind(exchange=exchange, queue=queue)
+
+    channel.basic_consume(queue=queue, on_message_callback=callback, auto_ack=True)
+
+    print('[^] WAITING')
+    channel.start_consuming()
+
+
+if __name__ == "__main__":
+    consume(callback=callback_1, queue=QUEUE, exchange=EXCHANGE)

balakhonov_danila_lab_4/program/consumer_2.py

@@ -0,0 +1,13 @@
+from consumer_1 import consume
+from publisher import EXCHANGE
+
+QUEUE = 'consumer_2_queue'
+
+def callback(ch, method, properties, body):
+    print(f'[x] CONSUMER 2: GET.. {body.decode()}')
+
+    print('[!] CONSUMER 1 OFF')
+
+
+if __name__ == "__main__":
+    consume(callback=callback, queue=QUEUE, exchange=EXCHANGE)

balakhonov_danila_lab_4/program/publisher.py

@@ -0,0 +1,30 @@
+import random
+import pika
+import time
+
+EXCHANGE = 'report_events'
+EVENTS = [
+    "Создать отчет трат",
+    "Создать отчет пополнений",
+    "Создать отчет смещения относительно плана расходов",
+]
+
+def publish(channel, exchange, event):
+    channel.basic_publish(exchange=exchange, routing_key='', body=event)
+    print(f'[x] PUBLISHED: {event}')
+    time.sleep(1)
+
+def run():
+    # Подключение
+    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
+    channel = connection.channel()
+    channel.exchange_declare(exchange=EXCHANGE, exchange_type='fanout')
+
+    # Публикация
+    while True:
+        event = EVENTS[random.randint(0, len(EVENTS) - 1)]
+        publish(channel, EXCHANGE, event)
+
+
+if __name__ == "__main__":
+    run()

balakhonov_danila_lab_4/tutor1/receive.py

@@ -0,0 +1,25 @@
+import pika, sys, os
+
+def main():
+    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
+    channel = connection.channel()
+
+    channel.queue_declare(queue='hello')
+
+    def callback(ch, method, properties, body):
+        print(f" [x] Received {body}")
+
+    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
+
+    print(' [*] Waiting for messages. To exit press CTRL+C')
+    channel.start_consuming()
+
+if __name__ == '__main__':
+    try:
+        main()
+    except KeyboardInterrupt:
+        print('Interrupted')
+        try:
+            sys.exit(0)
+        except SystemExit:
+            os._exit(0)

balakhonov_danila_lab_4/tutor1/send.py

@@ -0,0 +1,13 @@
+import pika
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
+channel = connection.channel()
+
+channel.queue_declare('hello')
+
+channel.basic_publish(exchange='',
+                      routing_key='hello',
+                      body='Hello world!')
+print(" [x] Sent 'Hello world!'")
+
+connection.close()

balakhonov_danila_lab_4/tutor2/new_task.py

@@ -0,0 +1,16 @@
+import pika
+import sys
+
+connection = pika.BlockingConnection(
+    pika.ConnectionParameters(host='localhost'))
+channel = connection.channel()
+
+channel.queue_declare(queue='task_queue', durable=True)
+
+message = ' '.join(sys.argv[1:]) or "Hello World!"
+channel.basic_publish(
+    exchange='',
+    routing_key='task_queue',
+    body=message)
+print(f" [x] Sent {message}")
+connection.close()

balakhonov_danila_lab_4/tutor2/worker.py

@@ -0,0 +1,22 @@
+import pika
+import time
+
+connection = pika.BlockingConnection(
+    pika.ConnectionParameters(host='localhost'))
+channel = connection.channel()
+
+channel.queue_declare(queue='task_queue', durable=True)
+print(' [*] Waiting for messages. To exit press CTRL+C')
+
+
+def callback(ch, method, properties, body):
+    print(f" [x] Received {body.decode()}")
+    time.sleep(body.count(b'.'))
+    print(" [x] Done")
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+channel.basic_qos(prefetch_count=1)
+channel.basic_consume(queue='task_queue', on_message_callback=callback)
+
+channel.start_consuming()

balakhonov_danila_lab_4/tutor3/emit_log.py

@@ -0,0 +1,13 @@
+import pika
+import sys
+
+connection = pika.BlockingConnection(
+    pika.ConnectionParameters(host='localhost'))
+channel = connection.channel()
+
+channel.exchange_declare(exchange='logs', exchange_type='fanout')
+
+message = ' '.join(sys.argv[1:]) or "info: Hello World!"
+channel.basic_publish(exchange='logs', routing_key='', body=message)
+print(f" [x] Sent {message}")
+connection.close()

balakhonov_danila_lab_4/tutor3/receive_logs.py

@@ -0,0 +1,22 @@
+import pika
+
+connection = pika.BlockingConnection(
+    pika.ConnectionParameters(host='localhost'))
+channel = connection.channel()
+
+channel.exchange_declare(exchange='logs', exchange_type='fanout')
+
+result = channel.queue_declare(queue='', exclusive=True)
+queue_name = result.method.queue
+
+channel.queue_bind(exchange='logs', queue=queue_name)
+
+print(' [*] Waiting for logs. To exit press CTRL+C')
+
+def callback(ch, method, properties, body):
+    print(f" [x] {body}")
+
+channel.basic_consume(
+    queue=queue_name, on_message_callback=callback, auto_ack=True)
+
+channel.start_consuming()