Merge pull request 'nikolaeva_yana_lab_5' (#339) from nikolaeva_yana_lab_5 into nikolaeva_yana_lab_4

Reviewed-on: #339

nikolaeva_yana_lab_4/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.9-slim
+
+RUN pip install pika
+
+WORKDIR /app
+COPY . /app
+
+CMD ["python", "publisher.py"]

nikolaeva_yana_lab_4/README.md

@@ -0,0 +1,21 @@
+### Лабораторная работа №4 - Работа с брокером сообщений
+
+#### Задание
+
+1. Установить брокер сообщений RabbitMQ.
+2. Пройти уроки 1, 2 и 3 из RabbitMQ Tutorials на любом языке программирования.
+3. Продемонстрировать работу брокера сообщений.
+
+#### Описание работы программы:
+
+- **Класс Publisher** успешно осуществляет отправку сообщений своим клиентам.
+  
+- **Класс Consumer1** принимает и обрабатывает сообщения с задержкой в 3 секунды, что можно заметить на видео.
+
+- **Класс Consumer2** мгновенно принимает и обрабатывает сообщения.
+
+### Скрины во вложениях
+
+## Видео 
+
+https://cloud.mail.ru/public/kzJ8/hUmC6959a

nikolaeva_yana_lab_4/consumer_1.py

@@ -0,0 +1,20 @@
+import pika
+import time
+
+def callback(ch, method, properties, body):
+    print(f" [Consumer 1] {body.decode('utf-8')}")
+    time.sleep(3)
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='rabbitmq'))
+channel = connection.channel()
+channel.exchange_declare(exchange='lunch_logs', exchange_type='fanout')
+
+queue_name = "lunch_queue_slow"
+channel.queue_declare(queue=queue_name)
+channel.queue_bind(exchange='lunch_logs', queue=queue_name)
+
+print(' [*] Consumer 1 waiting for logs. To exit press CTRL+C')
+
+channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=False)
+channel.start_consuming()

nikolaeva_yana_lab_4/consumer_2.py

@@ -0,0 +1,19 @@
+import pika
+
+def callback(ch, method, properties, body):
+    print(f" [Consumer 2] {body.decode('utf-8')}")
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='rabbitmq'))
+channel = connection.channel()
+channel.exchange_declare(exchange='lunch_logs', exchange_type='fanout')
+
+
+queue_name = "lunch_queue_fast"
+channel.queue_declare(queue=queue_name)
+channel.queue_bind(exchange='lunch_logs', queue=queue_name)
+
+print(' [*] Consumer 2 waiting for logs. To exit press CTRL+C')
+
+channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=False)
+channel.start_consuming()

nikolaeva_yana_lab_4/docker-compose.yml

@@ -0,0 +1,50 @@
+version: '3'
+
+services:
+  rabbitmq:
+    image: rabbitmq:3-management
+    container_name: rabbitmq_TEST
+    ports:
+      - "5672:5672"
+      - "15672:15672"
+    environment:
+      RABBITMQ_DEFAULT_USER: guest
+      RABBITMQ_DEFAULT_PASS: guest
+    healthcheck:
+      test: ["CMD", "rabbitmqctl", "status"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+
+  publisher:
+    build:
+      context: .
+    container_name: publisher
+    environment:
+      - PYTHONUNBUFFERED=1
+    command: python publisher.py
+    depends_on:
+      rabbitmq:
+        condition: service_healthy
+
+  consumer_1:
+    build:
+      context: .
+    container_name: consumer_1
+    environment:
+      - PYTHONUNBUFFERED=1
+    command: python consumer_1.py
+    depends_on:
+      rabbitmq:
+        condition: service_healthy
+
+  consumer_2:
+    build:
+      context: .
+    container_name: consumer_2
+    environment:
+      - PYTHONUNBUFFERED=1
+    command: python consumer_2.py
+    depends_on:
+      rabbitmq:
+        condition: service_healthy

nikolaeva_yana_lab_4/publisher.py

@@ -0,0 +1,20 @@
+import pika
+import time
+import random
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='rabbitmq'))
+channel = connection.channel()
+channel.exchange_declare(exchange='lunch_logs', exchange_type='fanout')
+
+events = [
+    "Новый заказ на завтрак",
+    "Новый заказ на обед",
+    "Новый заказ на ужин",
+    "Пользователь запросил меню"
+]
+
+while True:
+    message = random.choice(events)
+    channel.basic_publish(exchange='lunch_logs', routing_key='', body=message)
+    print(f" [x] Sent {message}")
+    time.sleep(1)

nikolaeva_yana_lab_4/tutorial_1/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)

nikolaeva_yana_lab_4/tutorial_1/send.py

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

nikolaeva_yana_lab_4/tutorial_2/new_task.py

@@ -0,0 +1,19 @@
+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,
+    properties=pika.BasicProperties(
+        delivery_mode=pika.DeliveryMode.Persistent
+    ))
+print(f" [x] Sent {message}")
+connection.close()

nikolaeva_yana_lab_4/tutorial_2/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()

nikolaeva_yana_lab_4/tutorial_3/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()

nikolaeva_yana_lab_4/tutorial_3/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()

nikolaeva_yana_lab_5/README.md

@@ -0,0 +1,62 @@
+# Лабораторная работа: Умножение матриц
+
+## Описание
+
+**Цель работы** – реализовать последовательный и параллельный алгоритмы умножения матриц, а также сравнить их производительность на больших квадратных матрицах.
+
+### Задачи:
+1. Реализовать последовательный алгоритм умножения матриц.
+2. Реализовать параллельный алгоритм, позволяющий задавать количество потоков вручную.
+3. Провести тесты на матрицах размером 100x100, 300x300 и 500x500.
+4. Сделать выводы о влиянии размеров матриц и количества потоков на производительность алгоритмов.
+
+## Теоретическое обоснование
+
+Умножение матриц — вычислительно сложная операция с асимптотической сложностью O(N³) для матриц размером N×N. 
+Для ускорения вычислений используется параллелизация, где каждая часть вычислений выполняется в отдельном потоке.
+Однако эффективность параллельного подхода зависит от размеров задачи и числа потоков.
+
+## Реализация
+
+1. **Последовательный алгоритм**:
+   - Выполняет вычисления поэлементно для каждой строки первой матрицы и каждого столбца второй.
+   - Этот алгоритм не использует дополнительные ресурсы, кроме одного потока, и подходит для небольших задач.
+
+2. **Параллельный алгоритм**:
+   - Делит строки первой матрицы на группы, каждая из которых обрабатывается в отдельном потоке.
+   - Реализован с использованием модуля `multiprocessing` для управления потоками.
+   - Число потоков задается вручную для возможности анализа производительности.
+
+## Результаты тестирования
+
+### Условия тестирования
+- Размеры матриц: 100x100, 300x300, 500x500.
+- Количество потоков: 1 (последовательное выполнение), 2, 4.
+- Диапазон значений элементов матриц: от 0 до 200.
+
+
+## Выводы
+
+1. **Последовательный алгоритм**:
+   - Подходит для матриц небольшого размера (100x100), где накладные расходы на параллелизацию превышают выигрыши от многопоточности.
+
+2. **Параллельный алгоритм**:
+   - Значительно ускоряет умножение матриц с увеличением их размера.
+   - Для матриц 500x500 ускорение в 2–2.5 раза при переходе от 1 потока к 4 потокам.
+
+3. **Влияние числа потоков**:
+   - Оптимальное число потоков зависит от размера задачи и доступных ресурсов.
+   - Слишком большое количество потоков может привести к росту накладных расходов.
+
+4. **Закономерности**:
+   - Накладные расходы на управление потоками минимальны для больших задач.
+   - Параллельные алгоритмы демонстрируют преимущество на задачах с высокой вычислительной сложностью.
+
+## Заключение
+
+Лабораторная работа подтвердила, что параллельные алгоритмы значительно эффективнее на больших данных.
+Однако для небольших задач последовательный алгоритм остается предпочтительным из-за отсутствия накладных расходов. 
+В реальных приложениях важно учитывать баланс между размером задачи и доступными вычислительными ресурсами.
+
+## Видео
+https://cloud.mail.ru/public/fykM/jy3KEZBZM

nikolaeva_yana_lab_5/main.py

@@ -0,0 +1,66 @@
+import numpy as np
+import time
+from multiprocessing import Pool, cpu_count
+
+
+def generate_matrix(size, value_range=(0, 200)):
+    return np.random.randint(value_range[0], value_range[1], (size, size))
+
+
+def multiply_matrices_sequential(matrix_a, matrix_b):
+    size = len(matrix_a)
+    result = np.zeros((size, size), dtype=int)
+    for i in range(size):
+        for j in range(size):
+            for k in range(size):
+                result[i][j] += matrix_a[i][k] * matrix_b[k][j]
+    return result
+
+
+def worker_multiply(args):
+    i_range, matrix_a, matrix_b, size = args
+    result_part = np.zeros((len(i_range), size), dtype=int)
+    for idx, i in enumerate(i_range):
+        for j in range(size):
+            for k in range(size):
+                result_part[idx][j] += matrix_a[i][k] * matrix_b[k][j]
+    return result_part
+
+
+def multiply_matrices_parallel(matrix_a, matrix_b, num_threads):
+    size = len(matrix_a)
+    step = size // num_threads
+    ranges = [range(i, min(i + step, size)) for i in range(0, size, step)]
+
+    with Pool(processes=num_threads) as pool:
+        results = pool.map(worker_multiply, [(i_range, matrix_a, matrix_b, size) for i_range in ranges])
+
+    return np.vstack(results)
+
+def benchmark(matrix_a, matrix_b, num_threads):
+    print(f"\nМатрицы размера {len(matrix_a)}x{len(matrix_a)}:")
+
+    start = time.time()
+    result_seq = multiply_matrices_sequential(matrix_a, matrix_b)
+    sequential_time = time.time() - start
+    print(f"Последовательное умножение заняло: {sequential_time:.4f} секунд")
+
+    start = time.time()
+    result_par = multiply_matrices_parallel(matrix_a, matrix_b, num_threads)
+    parallel_time = time.time() - start
+    print(f"Параллельное умножение ({num_threads} потоков) заняло: {parallel_time:.4f} секунд")
+
+    assert np.array_equal(result_seq, result_par), "Ошибка: результаты не совпадают!"
+    print("Результаты совпадают.")
+
+    return sequential_time, parallel_time
+
+
+if __name__ == "__main__":
+    sizes = [100, 300, 500]
+    num_threads = min(cpu_count(), 4)
+
+    for size in sizes:
+        matrix_a = generate_matrix(size)
+        matrix_b = generate_matrix(size)
+        benchmark(matrix_a, matrix_b, num_threads)