feat: 멀티 프로세싱으로 consume 병렬처리 구현

Author: ChaJunWoo-dev

src/main.py

@@ -1,17 +1,29 @@
 from dotenv import load_dotenv
 load_dotenv(override=True)
 
-import os
+from multiprocessing import Process
 
-from config.rabbitmq import connect
 from rabbitmq.consume import consume_message
+from config.rabbitmq import create_connection
+
+def worker():
+    connection, channel = create_connection()
+
+    channel.basic_qos(prefetch_count=1)
+    consume_message(channel)
+    connection.close()
 
 def main():
-    if not os.path.exists("temp"):
-        os.makedirs("temp")
+    workers = []
+
+    for _ in range(3):
+        p = Process(target=worker)
+
+        p.start()
+        workers.append(p)
 
-    connect()
-    consume_message()
+    for p in workers:
+        p.join()
 
 if __name__ == "__main__":
     main()

src/rabbitmq/consume.py

@@ -3,7 +3,6 @@
 
 from analyze_video import analyze_frame
 from config.constants import MESSAGES
-from config.rabbitmq import get_channel
 from gcs.generate_signed_url import generate_signed_url
 from gcs.read import get_video
 from gcs.write import upload_video
@@ -14,7 +13,7 @@
 from create_subtitle import make_stack_ass
 from insert_subtitle import insert_subtitle_to_video
 
-def callback(ch, method, properties, body):
+def process_message(body):
     try:
         email, file_name, selected_character = json.loads(body).values()
         edited_blob_name = f"edited/{file_name}"
@@ -31,13 +30,7 @@ def callback(ch, method, properties, body):
             for i, frame in enumerate(extract_frames(data_bytes)):
                 pose_data = analyze_frame(frame, side=selected_character)
 
-                label_frames(
-                    pose_data=pose_data,
-                    frame_id=i,
-                    sit_punch_frames=sit_punch_frames,
-                    uppercut_frames=uppercut_frames,
-                    hit_down_frames=hit_down_frames
-                )
+                label_frames(pose_data, i, sit_punch_frames, uppercut_frames, hit_down_frames)
 
             commands = get_commands(sit_punch_frames, uppercut_frames, hit_down_frames)
         except:
@@ -58,22 +51,25 @@ def callback(ch, method, properties, body):
             "message": MESSAGES.SUCCESS.ANALYZE,
             "url": signed_url
         }
+
+        return message
+
     except Exception as err:
         message = {
             "email": email,
             "message": str(err),
             "url": ""
         }
 
-        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+    return message
 
-    publish_message(message=message)
+def callback(ch, method, properties, body):
+    message = process_message(body)
+    publish_message(message)
 
     ch.basic_ack(delivery_tag=method.delivery_tag)
 
-def consume_message():
-    channel = get_channel()
-
+def consume_message(channel):
     channel.basic_consume(
         queue=os.environ["MQ_CONSUME_QUEUE"],
         auto_ack=False,