RabbitMQ

 

뉴스 기사 수집기 개발 단계에서 Scrapy를 통하여 수집한 기사를 MySQL 테이블에 적재하는 구조의 뉴스 기사 수집기이고 해당 단계에서 MySQL Connection Pool에 불필요한 커넥션이 쌓여 리소스를 낭비하는 문제가 발생하였습니다.

이에 Open source Message Queue 중 RabbitMQ를 통하여 MQ Broker를 구현하였습니다.

RabbitMQ는 docker를 이용하여 실행하였습니다.

 

 

 

https://www.rabbitmq.com/

Messaging that just works — RabbitMQ

 

얼랭(Erlang)으로 구현한 AMQP 프로토콜 기반의 브로커 시스템으로 오픈 소스이고 거의 모든 언어와 운영체제를 지원합니다.

   * 클라이언트 미들웨어 브로커 간 데이터 교환을 위한 MQ기반 메시지 교환 프로토콜

 

 

RabbitMQ의 기능

 

 

 

 

 

Producer.py

import pika

connection = pika.BlockingConnection(
pika.ConnectionParameters(host="localhost", port=29435)
)
channel = connection.channel()

channel.queue_declare(queue="qlist")

message = "daum success"
channel.basic_publish(
  exchange="",
  routing_key="qlist",
  body=message,
  properties=pika.BasicProperties(
  delivery_mode=2,  # make message persistent
  ),
)
connection.close()

Producer.py에서 queue 버퍼로 메시지를 전송합니다. 

pika 라이브러리를 통해 rabbitmq 채널에 연결하여 queue를 선언하여 body에 내용을 담아 해당 queue로 publish 하는 구조.

 

 

 

Consumer.py

import os
import sys

import pika


def main():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host="localhost", port=29435)
    )
    channel = connection.channel()

    channel.queue_declare(queue="qlist")

    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)

    channel.basic_consume(queue="qlist", on_message_callback=callback)

    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)

pika 라이브러리를 통해 채널에 커넥션하여 queue ready에 쌓여있는 message를 모두 send 합니다.

 

 

 

 

 

웹 UI화면에서 queue 진행 사항 확인 가능.