[MOM] RabbitMQ(3) - Producer 예제

• Producer와 Consumer에 대한 프로젝트를 구분하여 생성하고 공통적인 설정 부분부터 확인 후 각각의 소스를 처리하도록 하겠습니다.
  
• 먼저 메시지를 생성하는 Producer에 대해 알아보도록 하겠습니다.

---

#. build.grade

 

• RabbitMQ를 사용하기 위해선 라이브러리를 추가해야 합니다.
  
  
• 이는 Producer와 Consumer 모두에 공통적으로 적용됩니다.

implementation 'org.springframework.boot:spring-boot-starter-amqp'

---

#. application.yml

 

• username과 password의 경우 15672 포트에 연결했을 때 rabbitMQ에 접속하는 내용입니다.
  
  
• rabbitMQ의 guest 계정은 기본적으로 관리자 권한을 가지고 있습니다.
  
  
• Queue, exchange, routing에 대한 값을 편의에 따라 정의할 수 있으며 exchange 전략에 따라 추가/삭제할 수 있습니다.

server:
  port: 8032
spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    
rabbitmq:
  queue:
    name: sample-queue
  exchange:
    name: sample-exchange
  routing:
    key : key

---

#. Configuration

 

• RabbitMQ와 연결을 위한 설정 및 메시지 송신을 위한 Queue, Exchange, binding 등의 구성요소 설정입니다.
  
  
• 또한 RabbitMQ으로의 연결과 메시지 변환, 송신을 위한 Bean도 생성해야 합니다.
  
  
• bean에 대한 설명은 소스 내 주석을 참고해주시길 바랍니다.

@RequiredArgsConstructor
@Configuration
public class RabbitMqProducerConfig {

	@Value("${rabbitmq.queue.name}")
	private String queueName;
	
	@Value("${rabbitmq.exchange.name}")
	private String exchangeName;
	
	@Value("${rabbitmq.routing.key}")
	private String routingKey;
	
	@Value("${spring.rabbitmq.host}")
	private String host; 
	
	@Value("${spring.rabbitmq.port}")
	private int port;
	
	@Value("${spring.rabbitmq.username}")
	private String username;
	
	@Value("${spring.rabbitmq.password}")
	private String password;
	
	// 메시지의 송신 시 Queue를 정의하는 역할
	@Bean
	public Queue queue() {
		return new Queue(queueName, true);
	}
	
	// application.yml에서 지정한 Exchange 값으로 DirectExchange에 대한 Bean 생성
	@Bean
	public DirectExchange directExchange() {
		return new DirectExchange(exchangeName);
	}
	
	// application.yml에서 지정한 Queue 와 Exchange와 RoutingKey를 이용해 Binding 및 이에 대한 Bean 생성
	@Bean
	public Binding binding(Queue queue, DirectExchange directExchange) {
		return BindingBuilder.bind(queue).to(directExchange).with(routingKey);
	}
	
	// RabbitMQ 연동을 위한 ConnectionFactory Bean 생성
	@Bean
	public CachingConnectionFactory connectionFactory() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
		connectionFactory.setHost(host);
		connectionFactory.setPort(port);
		connectionFactory.setUsername(username);
		connectionFactory.setPassword(password);
		return connectionFactory;
	}
	
	//  ConnectionFactory로 연결 후 메시지를 주고 받는데 사용되는 인터페이스 Bean 생성
	// JSON과 메시지를 변환하기 위해 MessageConverter를 구성합니다.
	@Bean
	public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
		return rabbitTemplate;
	}
	
	// JSON 형식의 데이터를 직렬화 및 역직렬화하기 위해 사용되는 Converter Bean 생성
	// JSON을 개체로 송신
	@Bean
	public MessageConverter jackson2JsonMessageConverter() {
		return new Jackson2JsonMessageConverter();
	}
}

---

#. DTO

 

• 데이터를 송신하기 위한 DTO를 생성해줍니다.
  
  
• Lombok이 제대로 설치되었는지 다시 확인 바랍니다.
  
  
• 정상적으로 처리되지 않는게 있다면 롬복 jar파일을 더블클릭하여 설치해주시길 바랍니다.

@Data
public class MessageDto {
	private String title;
	private String content;
}

---

#. Controller

 

• HTTP 요청을 위한 Controller를 생성해줍니다.

@RequiredArgsConstructor
@RestController
public class RabbitMqProducerController {

	private final RabbitMqProducerService rabbitMqProducerService;
	
	@PostMapping("/send/message")
	public ResponseEntity<String> send(@RequestBody MessageDTO dto) {
		rabbitMqProducerService.sendMessage(dto);
		return ResponseEntity.ok("===== Send Complete !!  ====="); 
	}
}

---

#. Service

 

• RabbitTemplate을 이용해 메시지 변환 및 생성한 Queue로 전송하도록 합니다.

@Service
@RequiredArgsConstructor
public class RabbitMqProducerService {
	
	private final RabbitTemplate rabbitTemplate;
	
	@Value("${rabbitmq.exchange.name}")
	private String exchangeName;
	
	@Value("${rabbitmq.routing.key}")
	private String routingKey;

	public void sendMessage(MessageDTO dto) {
		rabbitTemplate.convertAndSend(exchangeName, routingKey, dto);
	}
}

---

• 위와 같이 로직을 만들어주고 localhost:15672 를 입력해 RabbitMQ UI로 접속해줍니다.
  
  
• ID 및 Password는 guest / guest 이며 신규 계정의 추가 및 권한부여도 가능합니다.

---

#. UI 메뉴 확인

• 접속하게 되면 아래와 같은 현황을 UI로 제공합니다.
  
  
• Overview 내 Queued messages를 보면 3가지 상태로 나뉘어져 있는 것을 볼 수 있습니다.

        1. Ready 
            - Consumer에게 아직 전달되지 않은 메시지의 수

        2. Unacked
            - Consumer에게 전달되었지만 아직 확인되지 않은 메시지 수
            - 일정 시간 내에 승인을 받지 못하면 메시지가 다른 소비자에게 다시 전달됩니다.

        3. Total  

            - 큐에 있는 전체 메시지 수

---

• 아래의 메뉴에서는 생성해준 Exchange와 Queue를 확인할 수 있습니다.

---

#. 메시지 전송

 

• Postman을 이용해 메시지를 보내보도록 하겠습니다.

---

• 이렇게 total 메시지 수에 1이 추가되었고, 아직 Consumer는 구현 전이기에 ready 상태로 돼있습니다.

---

• 한번 더 추가하면 2로 변하는 걸 볼 수 있습니다.

---

#. 이제 메시지를 수신하는 Consumer를 만들어보도록 하겠습니다.

 

 

===========================================================
틀린 내용이 있거나 이견 있으시면 언제든 가감 없이 말씀 부탁드립니다!
===========================================================