ALBERT SITE

오늘은 간단히 spring boot 로 RabbitMQ Producer 과 Consumer 만들어 보았다.

예전에는 rabbitMQ 공식 사이트상 제공해주는 jar 파일 넣어 단순 java 로 만들어서 운영하였는데 

srping boot 는 너무 간단히 구현 돼서 놀랐다.

1. maven 설정

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
</dependency>
<dependency>
   <groupId>org.springframework.amqp</groupId>
   <artifactId>spring-rabbit-test</artifactId>
   <scope>test</scope>
</dependency>

2. application.yml 내용 작성(기존 application.properties 삭제해도 되고 격식에 마춰 추가해줘도 상관 없음)

spring:
  rabbitmq:
    host: 192.168.200.160
    port: 5672
    username: rabbit
    password: s123456
    virtual-host:

3. 테스트 queue 생성

혹시나 ACCEPT_REFUSED 오류날시 해당 계정이 관련 VHOST 사용권한이 없어서 나타나는 거니 RABITMQ 서버 터미널상 아래와 같이 추가하여 주면 됨

[kafka@localhost .ssh]$ sudo rabbitmqctl set_permissions -p "/" "rabbit" ".*" ".*" ".*"

4. producer 만들기

package com.rbmq.example.hi;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class Producer {
    private static final Logger log = LoggerFactory.getLogger(Producer.class);

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Scheduled(fixedDelay = 2000, initialDelay = 1000)
    public void sendMessage(){
        rabbitTemplate.convertAndSend("hihi", "hello world2");
    }
}

@SpringBootApplication
@EnableScheduling
public class HiApplication {

   public static void main(String[] args) {
      SpringApplication.run(HiApplication.class, args);
   }

}

실행 후 rabbitmq 관리자 Queues 탭상 hihi 큐 totalCount 가 올라가는걸 확인 할 수 있다.

5. Consumer 만들기

package com.rbmq.example.hi;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    private static final Logger log = LoggerFactory.getLogger(Consumer.class);

    @RabbitListener(queues = "hihi")
    public void consume(Message message){
        log.info("consumer consumes message: {}",message);
    }
}

작성후 실행하면 조금전 hihi 큐가 싸였던 카운트 숫자가 줄어드는걸 볼수 있다.

6. 다중 채널의 컨슈머 만들기(채널은 말 그대로 동시에 일할 수 있는 일군을 늘린다는 뜻이고 일군이 많을 수록 동시에 더 많은 일을 처리할 수 있다.)

package com.rbmq.example.hi;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.ThreadLocalRandom;

@Component
public class MultipleConsumer {
    private static final Logger log = LoggerFactory.getLogger(MultipleConsumer.class);

    @RabbitListener(queues = "hihi", concurrency = "3")
    public void listen(String message){

        log.info("consume message {} on {} ", message, Thread.currentThread().getName());

        try {
            Thread.sleep(ThreadLocalRandom.current().nextLong(2000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

실행후 RabbitMQ관리자 Channels 탭을 확인하면 기존 1개의 채널이 있던 데로 부터 3개의 채널로 늘어난걸 확인할 수 있다.

실행 로그 확인하면 어떠한 내용이 몇번째 채널에서 실행됐는지도 확인 가능함

소스는 Github https://github.com/visionboy/RabbitMQ 에 올려 놓았음

끝