[RabbitMQ] rabbitMQ Spring Boot Producer, Consumer 생성, MultipleConsumer, prefetch

 

0.  RabbitMq 설치 및 설정

[RabbitMQ] rabbitmq 설치 및 설정, rabbitmq spring 3.0이상 연동

1. RabbitMqProducer 파일 생성

• yml 값 변수 매핑
• RabbitTemplate 선언

@Service
@Log4j2
public class RabbitMQProducer {

@Value("${rabbitmq.queue1.exchange}")
    private String exchange1;

    @Value("${rabbitmq.queue2.exchange}")
    private String exchange2;

    @Value("${rabbitmq.queue3.exchange}")
    private String exchange3;

    @Value("${rabbitmq.queue4.exchange}")
    private String emailExchange;


    @Value("${rabbitmq.routing.key.queue1}")
    private String routingKey1;

    @Value("${rabbitmq.routing.key.queue2}")
    private String routingKey2;

    @Value("${rabbitmq.routing.key.queue3}")
    private String routingKey3;

    @Value("${rabbitmq.routing.key.queue1}")
    private String emailRoutingKey1;

    @Value("${rabbitmq.routing.key.queue2}")
    private String emailRoutingKey2;

    private RabbitTemplate rabbitTemplate;
    
}

 

2. 각각의 큐의 Produce 메서드 생성

• rabbitTemplate.convertAndSend() 함수의 매개변수
  • 첫번째 : exchange 이름
  • 두번째 : routing key
  • 세번째 : 전달 Object

private RabbitTemplate rabbitTemplate;

public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
    this.rabbitTemplate = rabbitTemplate;
}

public void sendQueue1Message(SendingMsgDTO sendingMsgDTO){
    rabbitTemplate.convertAndSend(exchange1, routingKey1, sendingMsgDTO);

}
public void sendQueue2Message(SendingMsgDTO sendingMsgDTO){
    rabbitTemplate.convertAndSend(exchange2, routingKey2, sendingMsgDTO);
}

public void sendQueue3Message(SendingMsgDTO sendingMsgDTO){
    rabbitTemplate.convertAndSend(exchange3, routingKey3, sendingMsgDTO);
}

public void sendEmailQueue1Message(SendingEmailDTO sendingEmailDTO){
    rabbitTemplate.convertAndSend(emailExchange, emailRoutingKey1, sendingEmailDTO);
}

// 필자가 생성한 5개의 큐 중 2개의 큐는 exchange는 동일하며 routing key 가 다르므로 "emailExchange" 동일한 매개변수 전달
public void sendEmailQueue2Message(SendingEmailDTO sendingEmailDTO){
    rabbitTemplate.convertAndSend(emailExchange, emailRoutingKey2, sendingEmailDTO);
}

producer 호출 예시

@Service
@RequiredArgsConstructor
@Log4j2
public class SendMsgServiceImpl implements SendMsgService {

    @Autowired
    private final RabbitMQProducer rabbitMQProducer;

    public void produceQueue(SendingDTO sendingDTO, Map<Long, List<SendingMsgDTO>> map) {

        List<SendingMsgDTO> broker1SendingMsgDTOList = map.get(1L);
            broker1SendingMsgDTOList.forEach(sendingMsgDTO -> {
               rabbitMQProducer.sendQueue1MessagesendingMsgDTO);
            });
        });
    }    
}

3. application.yml RabbitMq Listener 설정 추가

prefetch

• Queue의 메시지를 Consumer의 메모리에 쌓아둘 수 있는 최대 메시지의 양 (spring default prefetch 값은 250)
• 만약 Prefetch = 250일 경우, RabbitMQ는 250개의 메시지까지 한번에 consumer의 메모리에 Push
• prefetch 값에 따라 성능 차이가 생기므로 어플리케이션에 맞게 조정을 해야함

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 50

3. Consumer 파일 생성

Multi Consume, concurrency

• Consumer를 멀티 스레드를 통해 사용하는 것
• 메시지 소비 속도가 빠름
• Producer의 메시지 발행이 Consumer의 소비보다 빠를 때 사용하면 성능이 좋음

@Component
public class Consumer {

    @Autowired
    private SendingService sendingService;
    
    @RabbitListener(queues = "${rabbitmq.routing.key.queue1}", concurrency = "10")
    public void consumeSKT(SendManagerMsgDTO sendManagerMsgDTO){
    
       // consume 후 프로세스 로직 
       sendingService.sendMsgProcessing(brokerId,new SendMsgProceessingDTO(1L,sendManagerMsgDTO)));
    }
}

 

 

참고

https://minholee93.tistory.com/entry/RabbitMQ-Prefetch

https://velog.io/@sdb016/RabbitMQ-Prefetch와-성능