0Pricing
AI Agents · 강의

이벤트 큐와 메시지 브로커

분리된 에이전트 통신을 위해 Redis 큐, RabbitMQ, Kafka를 활용합니다.

이벤트 큐와 메시지 브로커은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

에이전트에 메시지 대기열을 사용하는 이유

메시지 대기열은 트리거(이벤트 생성자)와 에이전트(이벤트 소비자)를 분리합니다. 생성자는 어떤 속도로든 이벤트를 발생시킬 수 있고, 에이전트는 자신의 처리 속도에 맞춰 이벤트를 처리합니다. 이를 통해 과부하를 방지하고 재시도를 활성화할 수 있습니다.

대기열로 사용하는 Redis 목록

Redis 목록은 간단한 대기열로 작동합니다. LPUSH는 앞쪽에 추가하고(대기열에 추가), BRPOP은 뒤쪽에서 제거하며 비어 있으면 대기합니다(대기열에서 제거). 이는 신뢰할 수 있는 FIFO 대기열입니다.

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)
QUEUE_NAME = 'agent:tasks'

# Producer: add a task to the queue
def enqueue_task(task: dict):
    import json
    r.lpush(QUEUE_NAME, json.dumps(task))
    print(f'Enqueued task: {task["id"]}')

# Consumer: blocking pop (waits up to 5 seconds for a message)
def dequeue_task(timeout: int = 5):
    import json
    result = r.brpop(QUEUE_NAME, timeout=timeout)
    if result:
        queue_name, raw_data = result
        return json.loads(raw_data)
    return None  # Timed out - no tasks

# Producer side
enqueue_task({'id': 'task-1', 'type': 'email', 'email_id': 'email-abc'})
enqueue_task({'id': 'task-2', 'type': 'email', 'email_id': 'email-def'})

# Check queue length
print(f'Queue length: {r.llen(QUEUE_NAME)}')

Redis를 사용하는 에이전트 작업자 루프

에이전트 작업자는 새 작업이 있는지 대기열을 계속 확인하고, 각 작업을 처리한 다음 반복합니다. 제한 시간 안에 작업이 도착하지 않으면 리소스를 소모하지 않고 다시 반복합니다.

import redis
import json
import time
import logging

logger = logging.getLogger('worker')
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def process_task(task: dict) -> bool:
    task_type = task.get('type')
    if task_type == 'email':
        logger.info(f'Processing email: {task.get("email_id")}')
        # Call email agent here
        return True
    elif task_type == 'file':
        logger.info(f'Processing file: {task.get("filepath")}')
        return True
    else:
        logger.warning(f'Unknown task type: {task_type}')
        return False

def run_worker(queue_name: str = 'agent:tasks'):
    print(f'Worker started, watching queue: {queue_name}')
    while True:
        try:
            task = dequeue_task(timeout=5)
            if task:
                success = process_task(task)
                if not success:
                    # Re-queue failed tasks for retry
                    r.lpush('agent:failed', json.dumps(task))
            else:
                logger.debug('No tasks, waiting...')
        except KeyboardInterrupt:
            print('Worker stopped')
            break
        except Exception as e:
            logger.error(f'Worker error: {e}')
            time.sleep(1)  # Brief pause on unexpected error

print('Worker function defined')

이벤트를 위한 Redis 게시/구독

Redis 게시/구독은 목록과 다릅니다. 현재 구독 중인 모든 대상에 메시지를 브로드캐스트합니다. 여러 에이전트가 같은 이벤트에 반응해야 하는 팬아웃 알림에는 게시/구독을 사용하십시오.

import redis
import json
import threading

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Publisher: send event to all subscribers
def publish_event(channel: str, event: dict):
    r.publish(channel, json.dumps(event))
    print(f'Published to {channel}: {event}')

# Subscriber: listen for events in a background thread
def subscribe_and_handle(channel: str, handler_fn):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    
    def listener():
        for message in pubsub.listen():
            if message['type'] == 'message':
                event = json.loads(message['data'])
                handler_fn(event)
    
    thread = threading.Thread(target=listener, daemon=True)
    thread.start()
    return thread

def handle_agent_event(event):
    print(f'Agent received event: {event}')

# Start subscriber
subscribe_and_handle('agent:events', handle_agent_event)

# Publish an event
publish_event('agent:events', {'type': 'user_action', 'action': 'login', 'user_id': 42})

import time
time.sleep(0.1)  # Give subscriber time to receive

pika를 사용한 RabbitMQ 기초

RabbitMQ는 경로 지정, 확인 응답, 배달 불능 메시지 대기열을 제공하는 모든 기능을 갖춘 메시지 브로커입니다. 피카 라이브러리는 파이썬을 RabbitMQ에 연결합니다.

import pika
import json

# Connect to RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host='localhost',
        port=5672,
        credentials=pika.PlainCredentials('guest', 'guest')
    )
)
channel = connection.channel()

# Declare a durable queue (survives broker restart)
channel.queue_declare(
    queue='agent_tasks',
    durable=True  # Queue survives RabbitMQ restart
)

# Publish a message
def publish_task(task: dict):
    channel.basic_publish(
        exchange='',
        routing_key='agent_tasks',
        body=json.dumps(task),
        properties=pika.BasicProperties(
            delivery_mode=2,  # Make message persistent
            content_type='application/json'
        )
    )
    print(f'Published task: {task["id"]}')

publish_task({'id': 'task-1', 'type': 'process_email', 'email_id': 'email-xyz'})
connection.close()

확인 응답을 사용하는 RabbitMQ 소비자

소비자는 처리 후 메시지를 반드시 확인 응답해야 합니다. 에이전트가 확인 응답을 보내기 전에 중단되면 RabbitMQ가 다른 소비자를 위해 메시지를 대기열에 다시 넣습니다. 이를 통해 메시지가 손실되지 않도록 합니다.

import pika
import json

def create_consumer():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='localhost')
    )
    channel = connection.channel()
    channel.queue_declare(queue='agent_tasks', durable=True)
    
    # Process one message at a time (fair dispatch)
    channel.basic_qos(prefetch_count=1)
    
    def process_message(ch, method, properties, body):
        task = json.loads(body)
        print(f'Processing: {task["id"]}')
        
        try:
            # Do agent work here
            print(f'Completed task: {task["id"]}')
            # Acknowledge: message is removed from queue
            ch.basic_ack(delivery_tag=method.delivery_tag)
        
        except Exception as e:
            print(f'Failed task {task["id"]}: {e}')
            # Negative acknowledge + requeue=True: puts message back
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    
    channel.basic_consume(queue='agent_tasks', on_message_callback=process_message)
    
    print('Consumer ready. Waiting for messages...')
    channel.start_consuming()

print('RabbitMQ consumer defined')

배달 불능 메시지 대기열

너무 여러 번 처리에 실패한 메시지는 영원히 재시도하지 말고 수동 검사를 위해 배달 불능 메시지 대기열(DLQ)로 보내야 합니다. 실패한 메시지가 자동으로 DLQ로 전달되도록 RabbitMQ를 구성하십시오.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Create the dead letter queue first
channel.queue_declare(queue='agent_tasks_dlq', durable=True)

# Create main queue with dead-letter exchange config
channel.queue_declare(
    queue='agent_tasks',
    durable=True,
    arguments={
        'x-dead-letter-exchange': '',
        'x-dead-letter-routing-key': 'agent_tasks_dlq',
        'x-message-ttl': 3600000,  # Messages expire after 1 hour
        'x-max-delivery-count': 3  # Max 3 delivery attempts
    }
)

# Consumer: nack without requeue sends to DLQ
def careful_consumer(ch, method, properties, body):
    import json
    task = json.loads(body)
    
    try:
        # Process task
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        # Do NOT requeue - send to DLQ
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

print('Dead letter queue configured')
connection.close()

작업 대기열 패턴

작업 대기열 패턴은 트리거와 실행을 분리합니다. 간단한 생성자가 작업 설명을 대기열에 넣고, 작업자들이 이를 가져와 실행합니다. 여러 작업자가 작업을 병렬로 처리할 수 있습니다.

import redis
import json
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Task descriptor - what needs to be done
def create_agent_task(task_type: str, params: dict, priority: int = 5) -> dict:
    return {
        'id': f'{task_type}-{int(datetime.utcnow().timestamp() * 1000)}',
        'type': task_type,
        'params': params,
        'priority': priority,
        'created_at': datetime.utcnow().isoformat(),
        'retry_count': 0,
        'max_retries': 3
    }

# Enqueue with priority (multiple queues by priority)
def enqueue_with_priority(task: dict):
    priority = task.get('priority', 5)
    queue = f'agent:tasks:p{priority}'
    r.lpush(queue, json.dumps(task))
    print(f'Enqueued {task["id"]} to priority-{priority} queue')

# Worker dequeues from high-priority queue first
def dequeue_priority(timeout: int = 5):
    for p in [1, 2, 3, 4, 5]:  # Check priority 1 first
        result = r.brpop(f'agent:tasks:p{p}', timeout=0.1)
        if result:
            return json.loads(result[1])
    return None

enqueue_with_priority(create_agent_task('email_analysis', {'email_id': 'abc'}, priority=2))
enqueue_with_priority(create_agent_task('file_process', {'path': '/tmp/file.txt'}, priority=5))

셀러리: 고수준 작업 대기열

셀러리는 가장 널리 사용되는 파이썬 작업 대기열입니다. 브로커로 Redis와 RabbitMQ를 지원하고, 자동 재시도, 작업 경로 지정, 플라워를 사용한 모니터링을 제공합니다. 운영 환경의 에이전트 작업에 사용하십시오.

from celery import Celery
import os

# Create Celery app with Redis broker
app = Celery(
    'agent_tasks',
    broker=os.environ.get('REDIS_URL', 'redis://localhost:6379/0'),
    backend=os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
)

# Configure retry behavior
app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    task_serializer='json',
    result_expires=3600
)

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def run_email_agent(self, email_id: str):
    try:
        # Agent logic here
        print(f'Processing email: {email_id}')
        return {'status': 'success', 'email_id': email_id}
    except Exception as exc:
        raise self.retry(exc=exc)

# Trigger a task (from any Python code)
# run_email_agent.delay('email-abc')  # Fire and forget
# result = run_email_agent.apply_async(args=['email-abc'], countdown=60)  # Delayed
print('Celery task defined')

대기열 상태 감시

대기열 깊이를 감시하여 적체를 감지하십시오. 대기열 깊이가 증가한다면 작업자가 더 필요하거나 에이전트가 너무 느린 것입니다. 깊이가 임계값을 초과하면 알림을 보내십시오.

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def check_queue_health(queue_name: str, max_depth: int = 100) -> dict:
    depth = r.llen(queue_name)
    oldest_raw = r.lindex(queue_name, -1)  # Get last item (oldest)
    
    oldest_age_seconds = None
    if oldest_raw:
        import json
        from datetime import datetime
        oldest = json.loads(oldest_raw)
        if 'created_at' in oldest:
            created = datetime.fromisoformat(oldest['created_at'])
            oldest_age_seconds = (datetime.utcnow() - created).total_seconds()
    
    health = {
        'queue': queue_name,
        'depth': depth,
        'max_depth': max_depth,
        'overloaded': depth > max_depth,
        'oldest_item_age_seconds': oldest_age_seconds
    }
    
    if health['overloaded']:
        print(f'ALERT: Queue {queue_name} depth={depth} exceeds max={max_depth}')
    
    return health

print('Queue health monitor defined')
print('Usage: check_queue_health("agent:tasks")')

Redis와 RabbitMQ 선택

단순성이 필요하거나 이미 Redis를 사용하고 있거나, 작업 수명이 짧고 중단 시 일부 메시지가 손실되어도 괜찮다면 Redis를 사용하십시오. 전달 보장, 복잡한 경로 지정, 배달 불능 메시지 대기열 또는 서로 다른 구독을 사용하는 여러 소비자가 필요하다면 RabbitMQ를 사용하십시오.

# Decision guide as code comments

# Use Redis lists when:
# - Simple FIFO queue is enough
# - Redis already in stack
# - Can tolerate rare message loss on Redis crash
# - Low operational overhead matters

# Use RabbitMQ when:
# - Message acknowledgment is critical (no data loss)
# - Need complex routing (topic exchanges, fanout)
# - Need dead-letter queues for failed messages
# - Multiple consumer types need different message subsets
# - Need message TTL and expiry

# Use Celery (on top of either) when:
# - Need task scheduling and delays
# - Need automatic retries with exponential backoff
# - Need task result storage
# - Need monitoring dashboard (Flower)

print('Redis: simple, fast, ok with rare loss')
print('RabbitMQ: guaranteed delivery, complex routing')
print('Celery: high-level abstraction over both')

지식 확인: 메시지 대기열

에이전트를 위한 메시지 대기열과 브로커에 대한 이해도를 확인해 보십시오.

메시지 대기열 요약

메시지 대기열은 에이전트 트리거와 실행을 분리합니다. Redis 목록은 LPUSH/BRPOP을 사용하는 간단한 FIFO 대기열을 제공하고, Redis 게시/구독은 브로드캐스트 이벤트를 활성화하며, RabbitMQ는 확인 응답과 배달 불능 메시지 대기열을 통해 전달을 보장합니다. 셀러리는 고수준 일정 관리, 재시도, 모니터링을 추가합니다. 대기열 깊이를 감시하면 적체가 서비스 중단으로 이어지기 전에 이를 알 수 있습니다.

자주 묻는 질문

“이벤트 큐와 메시지 브로커” 강의는 무료인가요?

네 — “이벤트 큐와 메시지 브로커” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“이벤트 큐와 메시지 브로커”에서 뭘 배우나요?

분리된 에이전트 통신을 위해 Redis 큐, RabbitMQ, Kafka를 활용합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.

“이벤트 큐와 메시지 브로커” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

  1. 에이전트 개발자를 위한 비동기 Python
  2. 이벤트 큐와 메시지 브로커
  3. 비차단 병렬 도구 실행
  4. 비동기 에이전트 프레임워크: LangChain과 그 너머
← AI Agents(으)로 돌아가기