DefStream-RabbitMQ

한국어 | English

DefStream에서 RabbitMQ를 사용할 수 있게 해주는 브로커 어댑터입니다. 일반 큐, Exchange 라우팅, 스트림 이력 읽기, 동기 RPC를 제공하며, DefStream Worker에 연결해 작업을 스레드·프로세스 풀에서 실행할 수 있습니다. 내부 클라이언트로 Pika를 사용합니다.

설치

Python 3.9 이상과 접속 가능한 RabbitMQ 서버가 필요합니다.

python -m pip install DefStream-RabbitMQ

DefStream>=0.1.0과 pika>=1.3.2도 함께 설치됩니다. 배포 패키지 이름은 DefStream-RabbitMQ이고 import 경로는 DefStream.Broker.Implements.RabbitMQ입니다.

큐에 보내고 받기

로컬 RabbitMQ의 guest 계정을 사용하는 예제입니다. 운영에서는 별도 계정과 환경변수·비밀 저장소를 사용하세요. 예제는 defstream.example.rabbitmq 큐를 만들며 실행 후에도 큐를 유지합니다.

from DefStream.Broker.Implements.RabbitMQ import Configuration, Connection


connection = Connection(Configuration(
  host='127.0.0.1',
  port=5672,
  username='guest',
  password='guest',
))
try:
  connection.connect()
  connection.create_queue('defstream.example.rabbitmq')
  queue = connection.queue('defstream.example.rabbitmq')
  queue.send({'message': 'Hello, DefStream!'})
  event = queue.pop(timeout=1000)
  if event is not None:
    print(event.body)
    event.ack()
finally:
  connection.close()

메시지는 기본 JSON 코덱으로 직렬화됩니다. pop() 제한 시간은 밀리초이며, 양수 제한 시간에 메시지가 없으면 None을 반환합니다. None 또는 0은 무기한 대기입니다. 처리가 성공하면 ack(), 재전달하려면 nack(), 거부하려면 reject()를 호출하세요. 거부한 메시지는 브로커의 dead-letter 설정이 있으면 해당 목적지로, 없으면 폐기됩니다.

연결 설정

timeout은 모든 작업에 대한 전체 실행 제한 시간이 아닙니다. 일반 연결의 heartbeat는 애플리케이션 작업·유휴 시간에 맞게 설정하고 처리되도록 관리하세요. 연결과 그 리소스는 소유 스레드에서만 사용해야 하며, 스레드·프로세스 사이에 공유하지 마세요.

Worker로 작업 처리하기

다음은 큐와 오류 큐를 준비한 후 작업을 소비하는 실행 파일입니다.

from signal import SIGINT, signal

from DefStream.Broker.Implements.RabbitMQ import (
  Configuration,
  Connection,
  Reject,
  Subscription,
)
from DefStream.Worker import Run, ThreadPool, Worker, WorkerSetup, declare


class Add(
  Run,
  parameters=Subscription(
    queue='defstream.example.rabbitmq.add',
    qos=4,
    error='defstream.example.rabbitmq.add.error',
  ),
  on_error=Reject(),
):
  def run(
    self,
    a: int,
    b: int,
  ):
    return a + b


class Prepare(WorkerSetup):
  def __call__(
    self,
    worker: Worker,
  ) -> None:
    for task in worker.tasks:
      declare(
        task=task,
        broker=worker.broker,
      )


if __name__ == '__main__':
  worker = Worker(
    Connection(Configuration(
      host='127.0.0.1',
      port=5672,
      username='guest',
      password='guest',
    )),
    ThreadPool(4),
    setups=Prepare(),
  )
  worker.add(Add)
  signal(
    SIGINT,
    lambda *_: worker.stop(),
  )
  worker.run()

다른 프로세스에서 같은 큐에 {'a': 10, 'b': 20}을 보내면 처리됩니다. Worker는 실행과 완료 훅이 끝난 뒤 ACK하고, 실패하면 지정한 정책을 적용합니다. on_error를 생략하면 기본값은 Requeue()입니다.

Subscription 옵션:

• queue: 작업 큐 이름.
• qos: 이 구독의 미확인 메시지 수 제한. 기본 소비자 값은 1이며, 0은 무제한입니다. Worker의 공유 풀 크기와 별개입니다.
• error: declare()가 생성하고 연결할 오류 큐 이름. 거부된 메시지의 DLQ입니다.
• offset: RabbitMQ 스트림 구독에 사용할 정수 오프셋.

Subscription 객체 생성이나 consumer.subs()는 큐를 선언하지 않습니다. 작업용 declare()는 일반 durable 큐와 선택적 DLQ를 준비하며, 스트림 큐나 Exchange 바인딩을 자동 생성하지 않습니다. 기존 큐의 선언 옵션이 다르면 실패하며 자동 삭제·재생성하지 않습니다.

동기 RPC

실행 중인 Worker가 있는 큐에 call()을 사용하면 Run.run()의 반환값을 받습니다. 아래 코드는 앞의 직접 연결 예제에서 connect()와 close() 사이에 사용하세요.

result = connection.queue('defstream.example.rabbitmq.add').call(
  {'a': 10, 'b': 20},
  timeout=5,
)
print(result)

• timeout은 초 단위이며 응답이 없으면 TimeoutError가 발생합니다. 이미 전달된 서버 작업을 취소하거나 전체 연결·정리 시간을 엄격히 제한하는 값은 아닙니다.
• 각 호출은 전용 연결과 임시 응답 큐를 사용하고 종료 시 정리합니다.
• 서버가 요청을 거부하면 DefStream.Broker.RemoteError가 발생합니다. 재큐잉은 최종 오류 응답 없이 재시도할 수 있습니다.
• 응답 발행과 요청 ACK는 원자적이지 않습니다. RPC에서도 중복 실행에 대비하세요.

공유 Run 클래스를 사용하는 애플리케이션은 코어의 request(body, task=Add, broker=connection, timeout=5)도 사용할 수 있습니다.

Exchange와 스트림

• connection.exchange(name)은 send()와 call()을 제공합니다. event=가 라우팅 키입니다. Exchange와 큐·바인딩을 먼저 선언하세요.
• Exchange RPC가 여러 큐로 라우팅되면 여러 곳에서 실행될 수 있으며, 클라이언트는 첫 일치 응답만 받습니다. 일반 RPC에는 단일 목적지를 사용하세요.
• create_exchange(), bind_exchange_to_queue(), bind_exchange_to_exchange()로 라우팅을 구성합니다. 각각에 대응하는 delete_... 또는 unbind_... 메서드도 있습니다.
• create_stream()으로 스트림 큐를 만든 뒤 connection.stream(name)의 send()와 read(offset=0, timeout=1000)을 사용합니다. 읽기 제한 시간은 밀리초입니다. 읽은 이벤트는 이터레이터가 다음 단계로 진행될 때 자동 ACK합니다.
• delete_queue(), delete_stream(), delete_exchange()는 리소스를 삭제하므로 애플리케이션 소유 리소스에만 사용하세요.

운영 시 주의사항

• 최소 한 번(at-least-once) 전달을 전제로 멱등 작업을 작성하세요.
• 무제한 재큐잉은 영구 오류 메시지의 반복 실행을 만들 수 있습니다. DLQ와 재시도 정책을 설계하세요.
• 소비자 연결에서 긴 작업을 직접 실행하지 말고 Worker의 풀을 사용하세요.
• 작업 훅에서 발행하려면 훅이 실행되는 스레드·프로세스 안에 별도 생산자 연결을 만드세요.
• worker.stop()은 처리 중인 작업과 ACK를 기다린 뒤 연결과 풀을 닫습니다. 강제 종료 또는 종료 대기 시간 제한은 제공하지 않습니다.

더 알아보기

• DefStream 사용 설명서: 작업, 풀, 훅, 검증, Inspector, 대시보드·콘솔
• 큐·Exchange 예제
• 독립 실행 RPC 예제
• 스트림 예제
• 릴리즈와 변경 내역

GNU GPL v3 라이선스를 사용합니다. 자세한 조건은 LICENSE를 확인하세요.