Spring Cloud #3 RabbitMQ

• docker로 RabbitMQ를 실행시킨다.

• 15672포트에서 RabbitMQ에서 접근이 가능하다

• ID, Password 모두 guest로 접속 가능하다.

 

Spring Project

• dependency 추가 (amqp)

dependencies {
    compileOnly("org.projectlombok:lombok")
    annotationProcessor("org.projectlombok:lombok")
    implementation("org.springframework.boot:spring-boot-starter-web")
    implementation("org.springframework.boot:spring-boot-starter-actuator")

    // rabbitMQ
    implementation("org.springframework.boot:spring-boot-starter-amqp")

}

• application.yml

server.port: 8081

spring:
  rabbitmq:
    addresses: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    cache:
      channel:
        size: 10
      connection:
        size: 10
      publisher-confirm: true

포트 설정 및 rabbitmq 설정

• RabbitMQ config

@Slf4j
@Configuration
@EnableRabbit
public class RabbitMQConfig {
    @Resource
    private RabbitProperties rabbitProperties;

    public static final String RABBIT_EXCHANGE_NAME = "tomato";

    @Bean
    public DefaultClassMapper classMapper() {
        DefaultClassMapper classMapper = new DefaultClassMapper();
        Map<String,Class<?>> idClassMapping = new HashMap<>();
        idClassMapping.put("myTask", MyTask.class);
        classMapper.setIdClassMapping(idClassMapping);
        return classMapper;
    }

    @Bean
    public MessageConverter rabbitMessageConverter(){
        Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter(MyJacksonConverter.getInstance());
        jsonConverter.setClassMapper(classMapper());
        return jsonConverter;
    }

    @Bean
    public ConnectionFactory rabbitConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        System.out.println("userName:: "+rabbitProperties.getUsername());
        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);

        return connectionFactory;
    }

    /**
     * Admin 설정 관련 부분
     * RabbitMQ에 admin 권한이 있는 계정으로 접속한 후에
     * exchange와 queue를 등록하고 매핑해준다.
     * @param rabbitConnectionFactory
     * @return
     */

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory rabbitConnectionFactory){
        final RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitConnectionFactory);
        // exchange 등록
        rabbitExchange(rabbitAdmin);
        // queue 자동 등록
        for(RabbitQueue rabbitQueue: RabbitQueue.values()){
            rabbitAdmin.declareQueue(new Queue(rabbitQueue.getQueueName(),true));
            rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(rabbitQueue.getQueueName(),true))
                    .to(rabbitExchange(rabbitAdmin)).with(rabbitQueue.getQueueName()));
        }
        rabbitAdmin.afterPropertiesSet();
        return rabbitAdmin;
    }

    @Bean
    TopicExchange rabbitExchange(RabbitAdmin rabbitAdmin) {
        TopicExchange topicExchange = new TopicExchange(RABBIT_EXCHANGE_NAME);
        topicExchange.setAdminsThatShouldDeclare(rabbitAdmin);
        return topicExchange;
    }
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory rabbitConnectionFactory,
                                         MessageConverter rabbitMessageConverter) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(rabbitConnectionFactory);
        rabbitTemplate.setMessageConverter(rabbitMessageConverter);
        rabbitTemplate.setExchange(RABBIT_EXCHANGE_NAME);

        rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            if(ack) {
                System.out.println("success");
            }else{
                System.out.println("error : "+cause);
            }
        }));

        return rabbitTemplate;
    }
}

• RabbitQueue 객체 custom

public enum RabbitQueue {
    SAMPLE_TASK("photo.sample"),
    READ("photo.read"),
    WRITE("photo.write"),
    EMPTY("photo.empty");

    private String queueName;

    RabbitQueue(String queueName) {
        this.queueName = queueName;
    }

    public String getQueueName(){return queueName;}

    public static RabbitQueue find(String name){
        for(RabbitQueue queue: RabbitQueue.values()){
            if(queue.getQueueName().equalsIgnoreCase(name)){
                return queue;
            }
        }
        return RabbitQueue.EMPTY;
    }
}

• RabbitMQ 메시지 publisher

@Component
public class RabbitMessagePublisher {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void publish(String routingKey, MyTask myTask){
        try{
            rabbitTemplate.convertAndSend("tomato", "photo.sample", myTask);
        }catch(Exception e){
            System.out.println("e = " + e);
        }
    }
}

• message publish

@RestController
public class PurchaseController {

    @Autowired
    RabbitMessagePublisher rabbitMessagePublisher;

    @GetMapping("/send")
    public String sendData(@RequestParam(defaultValue = "") String msg){
        rabbitMessagePublisher.publish(RabbitMQConfig.RABBIT_EXCHANGE_NAME, new MyTask(msg));
        return "send Data";
    }
}

• message receive 

@Component
public class RabbitMQReceiver {
    @RabbitListener(id = "photo.sample", queues = "photo.sample")
    public void handle(MyTask task){
        System.out.println("task = " + task);
    }
}

publish된 message를 Listener가 인식하여 메서드가 실행된다.