Spring 서버가 Producer 역할을 하도록 하는 방법을 설명한다. 그 중 direct exchange를 구현한다. RabbitMQ에 대한 설명은 아래 글에 적어두었다.

RabbitMQ 개념 알아보기

구현

1. docker-compose.xml

RabbitMQ 서버는 로컬에 container로 띄울 것이다.

  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    restart: unless-stopped
    ports:
      - "5672:5672"    # rabbitMQ 접속 포트
      - "15672:15672"  # rabbitMQ UI 접속 포트
    env_file:
      - .env.local
    environment:
      RABBITMQ_DEFAULT_USER: ${RABBITMQ_USERNAME}
      RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD}
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq

2. application.properties

RabbitMQ를 연결하는데 필요한 값들을 설정한다.

spring.rabbitmq.host=${RABBITMQ_HOST}
spring.rabbitmq.port=${RABBITMQ_PORT:5672}
spring.rabbitmq.username=${RABBITMQ_USERNAME}
spring.rabbitmq.password=${RABBITMQ_PASSWORD}

3. ConnectionFactory 생성

이렇게 설정을 하면 Spring이 RabbitProperties 객체를 생성하고, 위 값들을 읽어 아래처럼 바인딩한다.

@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitProperties

또한 Spring Boot 내부에 있는 RabbitAutoConfiguration 클래스가 다음과 같은 조건 검사를 하고 ,

@ConditionalOnClass(ConnectionFactory.class) // RabbitMQ 라이브러리 있으면
@ConditionalOnMissingBean(ConnectionFactory.class) // ConnectionFactory Bean 직접 안만들었으면

ConnectionFactory를 다음과 같은 코드로 자동 생성한다.

@Bean
public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties props) {
    CachingConnectionFactory cf = new CachingConnectionFactory();
    cf.setHost(props.getHost());
    cf.setPort(props.getPort());
    cf.setUsername(props.getUsername());
    cf.setPassword(props.getPassword());
    return cf;
}

4. RabbitMQ Topology 설정

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;

public class RabbitTopologyConfig {
    public static final String EXCHANGE = "diary.direct";
    public static final String ROUTING_KEY = "diary.analysis.request";
    public static final String QUEUE = "diary.analysis.queue";

    @Bean
    public Queue queue() {
		    // durable하게 만들어 큐를 디스크에 저장한다. RabbitMQ가 재시작해도 큐 유지
        return QueueBuilder.durable(QUEUE).build();
    }

    @Bean
    public DirectExchange exchange() {
		    // durable=true -> RabbitMQ 재시작해도 exchange 유지
		    // autoDelete=false -> consumer의 연결이 모두 끊어져도 exchange 유지
        return new DirectExchange(EXCHANGE, true, false);
    }

    @Bean
    Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

5. RabbitMQ Template 생성

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMqConfig {

		// Object -> JSON 변환. 없으면 byte[]로 보내서 Consumer가 파싱해야함.
    @Bean
    Jackson2JsonMessageConverter jacksonConverter(ObjectMapper mapper) {
        return new Jackson2JsonMessageConverter(mapper);
    }

		// Producer 클라이언트
    @Bean
    RabbitTemplate rabbitTemplate(
        ConnectionFactory cf, // Spring boot가 spring.rabbitmq.* 설정 보고 자동 생성
        Jackson2JsonMessageConverter converter
    ) {
        RabbitTemplate template = new RabbitTemplate(cf);
        template.setMessageConverter(converter); // JSON으로 변환
        
        // (선택) 다음과 같이 큐에 들어갔는지에 대한 ack을 받아볼 수 있다.
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("Publish to RabbitMQ success: analysis_id={}", correlationData.getId());
            } else {
                log.warn("Publish to RabbitMQ failed: analysis_id={}, cause={}", correlationData.getId(), cause);
            }
        });
        
        return template;
    }
}

6. publisher 생성

다음과 같이 publish를 수행하는 서비스를 만들 수 있다. CorrelationData 객체를 생성할 때 이벤트를 특정할 수 있는 id와 함께 넣어주어야 rabbitTemplate에서 ack에 대한 이벤트를 특정할 수 있다.

@Service
@RequiredArgsConstructor
public class AnalysisEventPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publish(Long diaryId, Long analysisId, String imageUrl) {

        DiaryAnalysisEvent diaryAnalysisEvent = new DiaryAnalysisEvent(diaryId, analysisId, imageUrl);

        rabbitTemplate.convertAndSend(
            RabbitTopologyConfig.EXCHANGE,
            RabbitTopologyConfig.ROUTING_KEY,
            diaryAnalysisEvent,
            new CorrelationData(String.valueOf(analysisId))
        );
    }
}

확인

localhost의 15672포트로 접근하면 작동하고 있는지 확인할 수 있다.