RabbitMQ의 Channel

com.rabbitmq.client.Channel 클래스는 RabbitMQ Java Client 라이브러리의 핵심 클래스이다.

이 클래스는 AMQP(Advanced Message Queuing Protocol) 채널을 나타내며, RabbitMQ는 메시지 큐 시스템으로, Channel은 애플리케이션과 RabbitMQ 서버 간의 통신을 처리하는 데 중요하다.

• 애플리케이션 <--> RabbitMQ의 대부분의 상호작용이 채널을 통해 이루어진다.

Channel의 주요 기능

1. 메시지 발행(publishing)

void basicPublish(
    String exchange, 
    String routingKey, 
    AMQP.BasicProperties props, 
    byte[] body)

Channel을 사용해서 메시지를 큐에 전송할 수 있다. 이때 특정 exchange와 routing key를 사용하여 발행한다.

2. 메시지 소비(consuming)

String basicConsume(
    String queue,
    boolean autoAck,
    Consumer callback)

지정된 큐에서 메시지를 소비한다. autoAck가 true면 자동으로 메시지 수신을 확인하고, 소비자 태그(cosumer tage)를 반환한다. 해당 메소드를 사용하면 메시지를 비동기적으로 소비할 수 있다.

3. 메시지 확인(acknowledgements) 🌟

메시지를 처리한 후 응답을 보내는 3가지 방식이 존재한다.

a. basicAck()

void basicAck(
	long deliveryTag,
	boolean multiple
) throws IOException;

// multiple = false (개별 ACK)
channel.basicAck(1, false); // 메시지 1 ACK
channel.basicAck(2, false); // 메시지 2 ACK
channel.basicAck(3, false); // 메시지 3 ACK
channel.basicAck(4, false); // 메시지 4 ACK
channel.basicAck(5, false); // 메시지 5 ACK

// multiple = true (Batch ACK)
channel.basicAck(5, true); // 메시지 1~5까지 한 번에 ACK

@RabbitListener(queues = "my-queue", concurrency = "5")
public void processMessage(String message) {
    // 멀티 스레드로 처리되므로 순서가 보장되지 않음
}

특정 메시지의 수신을 확인한다.

• deliveryTag: 확인(ACK)할 메시지의 고유 태그

• multiple:

  • true → 현재 deliveryTag까지 소비자가 받은 모든 메시지를 한꺼번에 ACK 처리

  • false → 해당 deliveryTag 메시지만 ACK 처리

b. basicNack()

void basicNack(
	long deliveryTag, 
	boolean multiple, 
	boolean requeue
) throws IOException;

channel.basicNack(deliveryTag, false, true); // 현재 메시지를 다시 큐로 보냄
channel.basicNack(deliveryTag, false, false); // 현재 메시지를 폐기

메시지의 수신을 거부하며, 리큐잉 여부를 결정할 수 있다.

• deliveryTag: 메시지 고유의 태그 (숫자 1, 2..)

• multiple:

  • true: 현재 deliveryTag까지 소비자가 받은 모든 메시지를 한꺼번에 Nack하는 것

  • false: 현재 메시지만 ack 처리

• requeue

  • true: 메시지를 다시 큐에 넣음 (재시도 가능)

  • false: 메시지를 버리거나 DLX로 보냄

c. basicReject()

void basicReject(
	long deliveryTag, 
	boolean requeue, 
) throws IOException;

basicNack과 유사하지만 단일 메시지의 수신을 거부한다.

• deliveryTag: 메시지 고유의 태그

• requeue:

  • true: 다시 queue에 넣음

  • false: 메시지를 버리거나 DLX로 보냄

4. 큐 관리(queue management)

a. Qeueu 선언

AMQP.Queue.DeclareOk queueDeclare(
    String queue, 
    boolean durable, 
    boolean exclusive, 
    boolean autoDelete, 
    Map<String, Object> arguments)

큐를 선언하는 메소드이다. 만약 큐가 존재하지 않으면 생성한다.

큐의 이름(name), 지속성(durable), 독점적(exclusive), 자동 삭제(autoDelete) 여부와 같은 속성을 설정할 수 있다.

• AMQP.Queue.DeclareOk와 같은 클래스는 RabbitMQ Client 라이브러리 내에서 AMQP 프로토콜의 규격을 따르는 클래스들 중 하나이다.

arguments에는 큐의 속성을 지정할 수 있다. 만약 큐에 ttl을 설정하면 메시지의 만료 시간을 설정할 수 있다.

Map<String, Object> arguments = new HashMap<>();
// 예시로 큐가 만료되는 시간 설정
arguments.put("x-message-ttl", 60000);  // 메시지 TTL: 60초

AMQP.Queue.DeclareOk declareOk = 
    channel.queueDeclare("hello", true, false, false, arguments);

b. Queue 삭제

AMQP.Queue.DeleteOk queueDelete(String queue)

지정한 이름의 큐를 삭제한다.

5. 익스체인지 관리(exchange mangement)

exchange는 메시지의 라우팅을 담당하는 역할을 하며, 큐와 직접적으로 연결되어 메시지를 큐에 전달한다.

AMQP.Exchange.DeclareOk exchangeDeclare(
    String exchange, 
    String type, 
    boolean durable)

• exchange의 이름

• type: direct, fanout, topic, headers 등

• durable: 서버 재시작 시에도 익스체인지가 유지될지 여부를 결정

6. 바인딩 관리(binding management)

queueBind는 큐와 익스체인지를 바인딩하는 데 사용된다. 바인딩이 이루어져야 메시지가 익스체인지에서 큐로 라우팅될 수 있게 된다.

AMQP.Queue.BindOk queueBind(
    String queue, 
    String exchange, 
    String routingKey)

• queue: 바인딩할 큐의 이름.

• exchange: 큐를 바인딩할 익스체인지의 이름.

• routingKey: 라우팅 키 (특히 direct, topic 익스체인지 타입에서 사용). fanout 타입의 익스체인지에서는 이 값을 무시한다.

설정을 통해 큐와 익스체인지를 연결하고 라우팅 규칙에 맞는 메시지가 해당 큐로 전달되도록 한다.

7. 흐름 제어(flow control)

이는 RabbitMQ의 QoS(Quality of Service) 설정을 위한 메서드로, 메시지를 소비하는 클라이언트가 한 번에 처리할 수 있는 메시지 수를 제한하는 데 사용된다. 즉, consumer가 한 번에 받을 수 있는 메시지 수를 제한한다.

이를 통해서 시스템이 과부화 되는 것을 방지하고, 메시지 처리를 효율적으로 관리하는 것을 목표로 한다.

void basicQos(int prefetchCount)

// 사용 예시
channel.queueDeclare("example-queue", true, false, false, null);

// 메시지를 한 번에 최대 1개만 소비하도록 설정
channel.basicQos(1);  // 한 번에 처리할 수 있는 메시지 수를 1로 설정

// 메시지 소비
System.out.println("Waiting for messages");

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println("Received: " + message);
    try {
        // 메시지 처리 시간 시뮬레이션
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        // 메시지 처리 완료 후 확인 응답
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
};

// 큐에서 메시지를 소비하고, `basicQos`에 의해 설정된 수만큼만 처리
channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {});

• prefetchCount: 한 번에 소비할 수 있는 최대 메시지 수를 지정한다. 'chunk'라고 생각하면 된다.

이 값을 설정하면 RabbitMQ 서버에서는 해당 수 만큼의 메시지를 한 번에 컨슈머에게 전달하고, 컨슈머가 응답을 보내기 전까지는 추가 메시지를 전달하지 않는다.

사용 예제

// 연결 생성
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();

// 채널 생성
Channel channel = connection.createChannel();

// 큐 선언
channel.queueDeclare("my-queue", true, false, false, null);

// 메시지 발행
String message = "Hello World!";
channel.basicPublish("", "my-queue", null, message.getBytes());

// 메시지 소비(수동 ACK 모드)
channel.basicConsume("my-queue", false, (consumerTag, delivery) -> {
    String receivedMessage = new String(delivery.getBody(), "UTF-8");
    System.out.println("Received: " + receivedMessage);
    
    // 메시지 처리 후 확인
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}, consumerTag -> {});

// 채널 및 연결 닫기
channel.close();
connection.close();

예제의 Spring AMQP에서는 저수준 API를 감싸고 있지만, @RabbitListenr의 ackMode = "MANUAL" 설정은 Channel 객체를 직접 다룰 필요가 있다.

다음은 Spring AMQP에서 Channel 객체를 사용하는 예제이다.

@RabbitListener(queues = "my-queue", ackMode = "MANUAL")
public void receive(Message message, Channel channel) throws IOException {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    try {
        // 메시지 처리
        String content = new String(message.getBody());
        System.out.println("Received: " + content);
        
        // 성공적으로 처리됨을 확인
        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        // 처리 실패 시 다시 큐에 넣음
        channel.basicNack(deliveryTag, false, true);
        // 또는 완전히 거부: channel.basicReject(deliveryTag, false);
    }
}

Channel 클래스는 RabbitMQ와의 통신을 위한 핵심 인터페이스이며, 메시지의 발행, 소비, 확인 및 큐 관리와 같은 모든 중요한 작업을 수행할 수 있다.