Messaging Queues

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

// 메시지 예시
{
  "header": {
    "messageId": "msg-123456",
    "timestamp": "2025-03-23T10:15:30Z",
    "priority": 1
  },
  "body": {
    "orderId": "ORD-9876",
    "customerId": "CUST-5432",
    "items": [
      {"productId": "PROD-001", "quantity": 2},
      {"productId": "PROD-015", "quantity": 1}
    ],
    "totalAmount": 79.98
  },
  "properties": {
    "routingKey": "orders.new",
    "contentType": "application/json",
    "expirationTime": 86400
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# RabbitMQ를 사용한 메시지 생산자 예시 (Python)
import pika
import json

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

# 큐 선언
channel.queue_declare(queue='order_processing', durable=True)

# 메시지 생성
message = {
    'orderId': 'ORD-9876',
    'customerId': 'CUST-5432',
    'items': [
        {'productId': 'PROD-001', 'quantity': 2},
        {'productId': 'PROD-015', 'quantity': 1}
    ],
    'totalAmount': 79.98
}

# 메시지 발행
channel.basic_publish(
    exchange='',
    routing_key='order_processing',
    body=json.dumps(message),
    properties=pika.BasicProperties(
        delivery_mode=2,  # 메시지 지속성 보장
        content_type='application/json'
    )
)

print(f"주문 처리 메시지 발행: {message['orderId']}")
connection.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# RabbitMQ를 사용한 메시지 소비자 예시 (Python)
import pika
import json
import time

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

# 큐 선언 (생산자와 동일한 큐)
channel.queue_declare(queue='order_processing', durable=True)

# 한 번에 하나의 메시지만 처리하도록 설정
channel.basic_qos(prefetch_count=1)

def callback(ch, method, properties, body):
    # 메시지 처리
    order = json.loads(body)
    print(f"주문 처리 중: {order['orderId']}")
    
    # 실제 비즈니스 로직 처리
    process_order(order)
    
    # 처리 완료 확인
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(f"주문 처리 완료: {order['orderId']}")

def process_order(order):
    # 주문 처리 로직 구현
    time.sleep(2)  # 처리 시간 시뮬레이션

# 메시지 소비 시작
channel.basic_consume(queue='order_processing', on_message_callback=callback)

print('주문 처리 서비스 실행 중... Ctrl+C로 종료')
channel.start_consuming()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

// Node.js Express API 엔드포인트 예시
const express = require('express');
const amqp = require('amqplib');
const app = express();

app.use(express.json());

// RabbitMQ 연결
let channel;
async function connectQueue() {
  const connection = await amqp.connect('amqp://localhost');
  channel = await connection.createChannel();
  await channel.assertQueue('image_processing', { durable: true });
}
connectQueue();

// 이미지 처리 API 엔드포인트
app.post('/api/images/process', async (req, res) => {
  const { imageUrl, filters, userId } = req.body;
  
  // 작업 ID 생성
  const jobId = generateJobId();
  
  // 작업을 큐에 추가
  channel.sendToQueue('image_processing', Buffer.from(JSON.stringify({
    jobId,
    imageUrl,
    filters,
    userId,
    timestamp: new Date().toISOString()
  })), { persistent: true });
  
  // 작업 ID와 상태 반환
  res.status(202).json({
    jobId,
    status: 'processing',
    statusUrl: `/api/jobs/${jobId}`
  });
});

// 작업 상태 확인 API 엔드포인트
app.get('/api/jobs/:jobId', async (req, res) => {
  const { jobId } = req.params;
  
  // 작업 상태 조회 로직
  const jobStatus = await getJobStatus(jobId);
  
  res.json(jobStatus);
});

app.listen(3000, () => {
  console.log('API 서버가 포트 3000에서 실행 중입니다.');
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// C# 이벤트 발행 서비스 예시
public class OrderService
{
    private readonly IMessageBroker _messageBroker;
    private readonly IOrderRepository _orderRepository;
    
    public OrderService(IMessageBroker messageBroker, IOrderRepository orderRepository)
    {
        _messageBroker = messageBroker;
        _orderRepository = orderRepository;
    }
    
    public async Task<Order> CreateOrderAsync(OrderRequest request)
    {
        // 주문 생성 및 저장
        var order = new Order
        {
            Id = Guid.NewGuid(),
            CustomerId = request.CustomerId,
            Items = request.Items,
            TotalAmount = request.Items.Sum(i => i.Price * i.Quantity),
            Status = OrderStatus.Created,
            CreatedAt = DateTime.UtcNow
        };
        
        await _orderRepository.SaveAsync(order);
        
        // 주문 생성 이벤트 발행
        await _messageBroker.PublishAsync("orders.created", new OrderCreatedEvent
        {
            OrderId = order.Id,
            CustomerId = order.CustomerId,
            TotalAmount = order.TotalAmount,
            Items = order.Items.Select(i => new OrderItemEvent
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity,
                Price = i.Price
            }).ToList(),
            CreatedAt = order.CreatedAt
        });
        
        return order;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// Java Spring Boot 요청-응답 패턴 예시
@Service
public class ProductService {
    
    private final RabbitTemplate rabbitTemplate;
    private final String requestQueue = "product.info.request";
    private final String replyQueue = "product.info.reply";
    
    public ProductService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public CompletableFuture<ProductInfo> getProductInfo(String productId) {
        CompletableFuture<ProductInfo> future = new CompletableFuture<>();
        
        String correlationId = UUID.randomUUID().toString();
        
        // 응답을 수신할 콜백 등록
        rabbitTemplate.convertAndSend(
            requestQueue,
            new ProductInfoRequest(productId),
            message -> {
                message.getMessageProperties().setReplyTo(replyQueue);
                message.getMessageProperties().setCorrelationId(correlationId);
                return message;
            }
        );
        
        // 비동기 응답 처리
        rabbitTemplate.receive(replyQueue, 30000, message -> {
            if (correlationId.equals(message.getMessageProperties().getCorrelationId())) {
                ProductInfo productInfo = (ProductInfo) rabbitTemplate.getMessageConverter()
                    .fromMessage(message);
                future.complete(productInfo);
            }
        });
        
        return future;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# Python FastAPI 경쟁 소비자 패턴 예시 (Celery 사용)
from fastapi import FastAPI, BackgroundTasks
from celery import Celery
from typing import List
import time

app = FastAPI()

# Celery 설정
celery_app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

# 데이터 처리 작업 정의
@celery_app.task
def process_data_chunk(chunk_id: int, data: List[dict]):
    print(f"Chunk {chunk_id} 처리 시작: {len(data)} 항목")
    # 실제 데이터 처리 로직
    time.sleep(len(data) * 0.1)  # 데이터 처리 시뮬레이션
    results = [item['value'] * 2 for item in data]
    print(f"Chunk {chunk_id} 처리 완료")
    return results

@app.post("/api/data/process")
async def process_large_dataset(data: List[dict], background_tasks: BackgroundTasks):
    # 대용량 데이터를 청크로 분할
    chunk_size = 100
    data_chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    
    # 각 청크를 별도 작업으로 큐에 추가
    task_ids = []
    for i, chunk in enumerate(data_chunks):
        task = process_data_chunk.delay(i, chunk)
        task_ids.append(task.id)
    
    return {
        "message": f"{len(data)} 항목이 {len(data_chunks)} 청크로 처리 중입니다.",
        "task_ids": task_ids,
        "status_url": "/api/tasks/status"
    }

@app.get("/api/tasks/status")
async def get_tasks_status(task_ids: List[str]):
    results = {}
    for task_id in task_ids:
        task = celery_app.AsyncResult(task_id)
        results[task_id] = {
            "status": task.status,
            "result": task.result if task.ready() else None
        }
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// Java에서 우선순위 큐 사용 예시
@Service
public class NotificationService {
    
    private final RabbitTemplate rabbitTemplate;
    
    @Autowired
    public NotificationService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void sendNotification(Notification notification) {
        // 알림 유형에 따라 우선순위 설정
        int priority = switch (notification.getType()) {
            case ALERT -> 10;  // 가장 높은 우선순위
            case WARNING -> 5;
            case INFO -> 1;    // 가장 낮은 우선순위
            default -> 1;
        };
        
        rabbitTemplate.convertAndSend(
            "notifications.exchange",
            "notifications.queue",
            notification,
            message -> {
                message.getMessageProperties().setPriority(priority);
                return message;
            }
        );
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

// Node.js에서 재시도 큐와 데드 레터 큐 구현 예시
const amqp = require('amqplib');

async function setupQueues() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  // 데드 레터 익스체인지 및 큐 설정
  await channel.assertExchange('dlx.payment', 'direct');
  await channel.assertQueue('payment.failed', {
    durable: true,
    arguments: {
      'x-message-ttl': 86400000  // 24시간 보관
    }
  });
  await channel.bindQueue('payment.failed', 'dlx.payment', 'payment');
  
  // 재시도 큐 설정
  await channel.assertQueue('payment.retry', {
    durable: true,
    arguments: {
      'x-dead-letter-exchange': '',
      'x-dead-letter-routing-key': 'payment.process',
      'x-message-ttl': 60000  // 1분 후 재시도
    }
  });
  
  // 메인 처리 큐 설정
  await channel.assertQueue('payment.process', {
    durable: true,
    arguments: {
      'x-dead-letter-exchange': 'dlx.payment',
      'x-dead-letter-routing-key': 'payment'
    }
  });
  
  return { connection, channel };
}

async function setupConsumer(channel) {
  channel.prefetch(1);
  
  channel.consume('payment.process', async (msg) => {
    try {
      const payment = JSON.parse(msg.content.toString());
      console.log(`결제 처리 중: ${payment.id}`);
      
      // 결제 처리 로직
      const processed = await processPayment(payment);
      
      if (processed) {
        // 성공적으로 처리됨
        channel.ack(msg);
        console.log(`결제 성공: ${payment.id}`);
      } else {
        // 일시적 오류, 재시도 필요
        const retryCount = (msg.properties.headers['x-retry-count'] || 0) + 1;
        
        if (retryCount <= 3) {
          // 재시도 큐로 전송
          channel.publish('', 'payment.retry', msg.content, {
            persistent: true,
            headers: { 'x-retry-count': retryCount }
          });
          channel.ack(msg);
          console.log(`결제 재시도 예약 (${retryCount}/3): ${payment.id}`);
        } else {
          // 최대 재시도 횟수 초과, 데드 레터 큐로 이동
          channel.nack(msg, false, false);
          console.log(`결제 실패, 데드 레터로 이동: ${payment.id}`);
        }
      }
    } catch (error) {
      // 처리 중 오류 발생
      console.error(`결제 처리 오류: ${error.message}`);
      channel.nack(msg, false, false);
    }
  });
}

function processPayment(payment) {
  // 실제 결제 처리 로직
  return Math.random() > 0.3;  // 70% 성공률 시뮬레이션
}

async function main() {
  const { connection, channel } = await setupQueues();
  await setupConsumer(channel);
  console.log('결제 처리 서비스 실행 중...');
}

main().catch(console.error);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

// C#에서 메시지 TTL 설정 예시 (RabbitMQ.Client 사용)
using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;

public class NotificationSender
{
    private readonly IModel _channel;
    
    public NotificationSender(IModel channel)
    {
        _channel = channel;
        
        // TTL이 적용된 큐 선언
        var queueArgs = new Dictionary<string, object>
        {
            { "x-message-ttl", 300000 }  // 5분 TTL
        };
        
        _channel.QueueDeclare(
            queue: "notifications.transient",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: queueArgs
        );
    }
    
    public void SendTransientNotification(string userId, string message)
    {
        var notification = new
        {
            UserId = userId,
            Message = message,
            Timestamp = DateTime.UtcNow
        };
        
        var body = Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(notification));
        
        var properties = _channel.CreateBasicProperties();
        properties.Persistent = true;
        
        _channel.BasicPublish(
            exchange: "",
            routingKey: "notifications.transient",
            basicProperties: properties,
            body: body
        );
    }
    
    public void SendPriorityNotification(string userId, string message, int expirationSeconds)
    {
        var notification = new
        {
            UserId = userId,
            Message = message,
            Timestamp = DateTime.UtcNow
        };
        
        var body = Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(notification));
        
        var properties = _channel.CreateBasicProperties();
        properties.Persistent = true;
        properties.Expiration = (expirationSeconds * 1000).ToString();  // 개별 메시지 TTL 설정
        
        _channel.BasicPublish(
            exchange: "",
            routingKey: "notifications.priority",
            basicProperties: properties,
            body: body
        );
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

// Java에서 지연 메시지 구현 예시 (Spring AMQP 사용)
@Configuration
public class RabbitMQConfig {
    
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("delay.exchange", "x-delayed-message", true, false, args);
    }
    
    @Bean
    public Queue reminderQueue() {
        return new Queue("reminders.queue", true);
    }
    
    @Bean
    public Binding reminderBinding(Queue reminderQueue, CustomExchange delayExchange) {
        return BindingBuilder.bind(reminderQueue)
                .to(delayExchange)
                .with("reminders.routing")
                .noargs();
    }
}

@Service
public class ReminderService {
    
    private final RabbitTemplate rabbitTemplate;
    
    @Autowired
    public ReminderService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void scheduleReminder(String userId, String message, long delayInSeconds) {
        Reminder reminder = new Reminder(
            UUID.randomUUID().toString(),
            userId,
            message,
            LocalDateTime.now().plusSeconds(delayInSeconds)
        );
        
        rabbitTemplate.convertAndSend(
            "delay.exchange",
            "reminders.routing",
            reminder,
            message -> {
                message.getMessageProperties().setHeader("x-delay", delayInSeconds * 1000);
                return message;
            }
        );
        
        System.out.println("알림 예약됨: " + delayInSeconds + "초 후 " + userId + "님에게 전송");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

// Kotlin Spring Boot 비동기 작업 처리 예시
@RestController
@RequestMapping("/api/reports")
class ReportController(
    private val reportPublisher: ReportPublisher,
    private val reportRepository: ReportRepository
) {
    
    @PostMapping
    fun generateReport(@RequestBody request: ReportRequest): ResponseEntity<ReportResponse> {
        // 보고서 작업 생성
        val reportJob = ReportJob(
            id = UUID.randomUUID().toString(),
            userId = request.userId,
            parameters = request.parameters,
            status = ReportStatus.PENDING,
            createdAt = LocalDateTime.now()
        )
        
        // 작업 저장
        reportRepository.save(reportJob)
        
        // 작업을 큐에 발행
        reportPublisher.publishReportJob(reportJob)
        
        // 즉시 응답 반환
        return ResponseEntity
            .accepted()
            .body(ReportResponse(
                jobId = reportJob.id,
                status = reportJob.status,
                statusUrl = "/api/reports/${reportJob.id}"
            ))
    }
    
    @GetMapping("/{jobId}")
    fun getReportStatus(@PathVariable jobId: String): ResponseEntity<ReportJobStatus> {
        val reportJob = reportRepository.findById(jobId)
            ?: return ResponseEntity.notFound().build()
            
        return ResponseEntity.ok(
            ReportJobStatus(
                jobId = reportJob.id,
                status = reportJob.status,
                progress = reportJob.progress,
                result = reportJob.result,
                createdAt = reportJob.createdAt,
                completedAt = reportJob.completedAt
            )
        )
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

// 주문 시스템에서 결제 시스템으로 메시지 전송 예시 (Java)
@Service
public class OrderCompletionService {
    
    private final JmsTemplate jmsTemplate;
    private final OrderRepository orderRepository;
    
    @Autowired
    public OrderCompletionService(JmsTemplate jmsTemplate, OrderRepository orderRepository) {
        this.jmsTemplate = jmsTemplate;
        this.orderRepository = orderRepository;
    }
    
    @Transactional
    public void completeOrder(String orderId, PaymentDetails paymentDetails) {
        // 주문 상태 업데이트
        Order order = orderRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));
            
        order.setStatus(OrderStatus.PAID);
        order.setPaymentDetails(paymentDetails);
        order.setUpdatedAt(LocalDateTime.now());
        
        orderRepository.save(order);
        
        // 결제 완료 메시지를 다른 시스템에 전송
        OrderPaidEvent event = new OrderPaidEvent(
            orderId,
            order.getCustomerId(),
            paymentDetails.getAmount(),
            paymentDetails.getPaymentMethod(),
            order.getItems(),
            order.getShippingAddress(),
            LocalDateTime.now()
        );
        
        // 배송 시스템에 알림
        jmsTemplate.convertAndSend("shipping.orders.paid", event);
        
        // 재고 시스템에 알림
        jmsTemplate.convertAndSend("inventory.orders.paid", event);
        
        // 고객 알림 시스템에 알림
        jmsTemplate.convertAndSend("notifications.orders.paid", event);
        
        System.out.println("주문 " + orderId + " 완료 메시지 전송됨");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

# Python FastAPI에서 이미지 처리 부하 분산 예시
from fastapi import FastAPI, File, UploadFile, BackgroundTasks
from fastapi.responses import JSONResponse
import aio_pika
import asyncio
import uuid
import json
import os

app = FastAPI()

# RabbitMQ 연결 설정
async def get_rabbitmq_connection():
    return await aio_pika.connect_robust("amqp://guest:guest@localhost/")

# 이미지 업로드 및 처리 요청 API
@app.post("/api/images/process")
async def process_image(file: UploadFile = File(...), background_tasks: BackgroundTasks):
    # 파일 저장
    file_id = str(uuid.uuid4())
    file_path = f"uploads/{file_id}_{file.filename}"
    os.makedirs("uploads", exist_ok=True)
    
    with open(file_path, "wb") as buffer:
        content = await file.read()
        buffer.write(content)
    
    # 이미지 처리 작업 큐에 전송
    background_tasks.add_task(send_to_processing_queue, file_id, file_path)
    
    return JSONResponse(
        status_code=202,
        content={
            "message": "이미지 처리가 시작되었습니다.",
            "file_id": file_id,
            "status_url": f"/api/images/status/{file_id}"
        }
    )

# 이미지 처리 작업을 큐에 전송
async def send_to_processing_queue(file_id: str, file_path: str):
    connection = await get_rabbitmq_connection()
    async with connection:
        channel = await connection.channel()
        
        # 이미지 처리 큐 선언
        queue = await channel.declare_queue(
            "image_processing",
            durable=True
        )
        
        # 메시지 생성 및 전송
        message_body = json.dumps({
            "file_id": file_id,
            "file_path": file_path,
            "timestamp": str(datetime.datetime.now())
        }).encode()
        
        await channel.default_exchange.publish(
            aio_pika.Message(
                body=message_body,
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT
            ),
            routing_key="image_processing"
        )
        
        print(f"이미지 처리 요청이 큐에 전송됨: {file_id}")

# 이미지 처리 상태 확인 API
@app.get("/api/images/status/{file_id}")
async def get_image_status(file_id: str):
    # 실제 구현에서는 DB에서 처리 상태 조회
    # 여기서는 간단한 예시만 표시
    return {
        "file_id": file_id,
        "status": "processing",  # 실제로는 DB에서 현재 상태 조회
        "message": "이미지가 처리 중입니다."
    }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226

// TypeScript에서 장애 복원력 패턴 구현 예시
import { Channel, Connection, connect } from 'amqplib';

class ResilientEmailService {
  private connection: Connection | null = null;
  private channel: Channel | null = null;
  private readonly maxRetries = 3;
  private isConnecting = false;
  private pendingMessages: any[] = [];

  // 연결 설정
  async setupConnection(): Promise<void> {
    if (this.isConnecting) return;
    
    this.isConnecting = true;
    
    try {
      // RabbitMQ 연결
      this.connection = await connect('amqp://localhost');
      
      // 연결 오류 처리
      this.connection.on('error', (err) => {
        console.error('RabbitMQ 연결 오류:', err);
        this.resetConnection();
      });
      
      this.connection.on('close', () => {
        console.log('RabbitMQ 연결이 닫혔습니다. 재연결 시도 중...');
        this.resetConnection();
        setTimeout(() => this.setupConnection(), 5000);
      });
      
      // 채널 생성
      this.channel = await this.connection.createChannel();
      
      // 이메일 큐 설정
      await this.channel.assertQueue('emails', { durable: true });
      
      // 재시도 큐 설정
      await this.channel.assertQueue('emails.retry', {
        durable: true,
        arguments: {
          'x-dead-letter-exchange': '',
          'x-dead-letter-routing-key': 'emails',
          'x-message-ttl': 60000  // 1분 후 재시도
        }
      });
      
      // 데드 레터 큐 설정
      await this.channel.assertQueue('emails.failed', { durable: true });
      
      console.log('RabbitMQ 연결 및 채널 설정 완료');
      
      // 대기 중인 메시지 처리
      this.processPendingMessages();
    } catch (error) {
      console.error('RabbitMQ 연결 설정 실패:', error);
      this.resetConnection();
      setTimeout(() => this.setupConnection(), 5000);
    } finally {
      this.isConnecting = false;
    }
  }
  
  // 연결 리셋
  private resetConnection(): void {
    this.channel = null;
    this.connection = null;
  }
  
  // 대기 중인 메시지 처리
  private async processPendingMessages(): Promise<void> {
    if (!this.channel || this.pendingMessages.length === 0) return;
    
    const messages = [...this.pendingMessages];
    this.pendingMessages = [];
    
    for (const msg of messages) {
      await this.sendEmailToQueue(msg.email, msg.retryCount);
    }
  }
  
  // 이메일 메시지를 큐에 전송
  async sendEmail(to: string, subject: string, body: string): Promise<void> {
    const email = { to, subject, body, timestamp: new Date().toISOString() };
    
    try {
      if (!this.channel) {
        await this.setupConnection();
        
        if (!this.channel) {
          // 여전히 연결이 없는 경우 대기 중인 메시지에 추가
          this.pendingMessages.push({ email, retryCount: 0 });
          return;
        }
      }
      
      await this.sendEmailToQueue(email, 0);
    } catch (error) {
      console.error('이메일 전송 오류:', error);
      // 오류 발생 시 대기 중인 메시지에 추가
      this.pendingMessages.push({ email, retryCount: 0 });
      
      // 연결 재설정 시도
      this.resetConnection();
      setTimeout(() => this.setupConnection(), 5000);
    }
  }
  
  // 이메일을 큐에 실제로 전송
  private async sendEmailToQueue(email: any, retryCount: number): Promise<void> {
    if (!this.channel) throw new Error('RabbitMQ 채널이 설정되지 않았습니다.');
    
    const message = {
      ...email,
      retryCount,
      sentAt: new Date().toISOString()
    };
    
    // 메시지를 이메일 큐에 발행
    this.channel.sendToQueue(
      'emails',
      Buffer.from(JSON.stringify(message)),
      {
        persistent: true,
        headers: { 'x-retry-count': retryCount }
      }
    );
    
    console.log(`이메일이 큐에 전송됨: ${email.to}, 재시도: ${retryCount}`);
  }
  
  // 이메일 소비자 설정
  async setupConsumer(): Promise<void> {
    if (!this.channel) {
      await this.setupConnection();
      if (!this.channel) throw new Error('RabbitMQ 채널이 설정되지 않았습니다.');
    }
    
    // 이메일 처리를 위한 소비자 설정
    this.channel.prefetch(1);
    this.channel.consume('emails', async (msg) => {
      if (!msg) return;
      
      try {
        const email = JSON.parse(msg.content.toString());
        console.log(`이메일 처리 중: ${email.to}`);
        
        // 실제 이메일 전송 로직
        const success = await this.deliverEmail(email);
        
        if (success) {
          // 성공적으로 처리됨
          this.channel?.ack(msg);
          console.log(`이메일 전송 성공: ${email.to}`);
        } else {
          // 전송 실패, 재시도 필요
          const retryCount = (msg.properties.headers['x-retry-count'] || 0) + 1;
          
          if (retryCount <= this.maxRetries) {
            // 재시도 큐로 전송
            this.channel?.publish(
              '',
              'emails.retry',
              msg.content,
              {
                persistent: true,
                headers: { 'x-retry-count': retryCount }
              }
            );
            this.channel?.ack(msg);
            console.log(`이메일 재시도 예약 (${retryCount}/${this.maxRetries}): ${email.to}`);
          } else {
            // 최대 재시도 횟수 초과, 실패 큐로 이동
            this.channel?.publish('', 'emails.failed', msg.content, { persistent: true });
            this.channel?.ack(msg);
            console.log(`이메일 전송 실패, 데드 레터로 이동: ${email.to}`);
          }
        }
      } catch (error) {
        console.error('이메일 처리 중 오류:', error);
        // 처리 오류, 데드 레터 큐로 이동
        this.channel?.nack(msg, false, false);
      }
    });
    
    console.log('이메일 소비자가 시작되었습니다.');
  }
  
  // 실제 이메일 전송 로직 (외부 서비스 호출)
  private async deliverEmail(email: any): Promise<boolean> {
    try {
      // 실제로는 SMTP 서비스 등을 호출
      console.log(`이메일 전송 시도: ${email.to}, 제목: ${email.subject}`);
      
      // 성공/실패 시뮬레이션 (80% 성공률)
      const isSuccessful = Math.random() > 0.2;
      
      if (isSuccessful) {
        return true;
      } else {
        console.log(`이메일 전송 일시적 실패: ${email.to}`);
        return false;
      }
    } catch (error) {
      console.error(`이메일 전송 중 오류: ${error}`);
      return false;
    }
  }
}

// 사용 예시
async function main() {
  const emailService = new ResilientEmailService();
  await emailService.setupConnection();
  await emailService.setupConsumer();
  
  // 이메일 전송 테스트
  await emailService.sendEmail(
    'user@example.com',
    '중요 알림',
    '귀하의 계정에 중요한 변경사항이 있습니다.'
  );
}

main().catch(console.error);