Springboot集成RabbitMQ【Direct Exchange】

Springboot集成RabbitMQ【Direct Exchange】

本文只针对路由直连模式（Direct Exchange）的使用。

---

一、生产者端的队列/交换器配置：
这里只有生产者客户端需要配置，消费者端可无需配置。

@Configuration
public class DirectRabbitConfig {

    
    @Bean
    public Queue CustomDirectQueue() {
    	// 这里Queue对象有四个常用参数
        // name：消息队列命名。
        // durable：默认为false，是否持久化。持久化队列会被存储在磁盘上，当消息代理重启时仍然存在；否则当前连接有效。
        // exclusive：默认为false，只能被当前创建的连接使用，当连接关闭后队列即被删除，优先级高于durable。
        // autoDelete：默认为false，是否自动删除，当没有生产者或者消费者使用此队列，则队列自动删除。
        return new Queue("CustomDirectQueue", true);
    }

    
    @Bean
    public DirectExchange CustomDirectExchange() {
        return new DirectExchange("CustomDirectExchange", true, false);
    }

    
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(CustomDirectQueue()).to(CustomDirectExchange()).with("custom.direct.routing");
    }

}

---

二、消费者端的监听消费：

注解@RabbitListener可标注在方法上，表示当监听到指定的队列中有消息时会由该方法进行接收处理。注解@RabbitListener也可标注在类上，但需要搭配注解@RabbitHandler使用，表示当监听到指定的队列中有消息时，就会交给对应的@RabbitHandler进行接收处理。注解@RabbitHandler可以有多个，但参数类型不能一样。交由哪个方法处理则根据MessageConverter转换后的参数类型来决定。

@Component
@RabbitListener(queues = "CustomDirectQueue")
public class DirectConsumer {

    @RabbitHandler
    public void process(Map message) {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("[consumer-1]从队列 [CustomDirectQueue] 收到消息：" + message.toString());
    }

    @RabbitListener(queues = "CustomDirectQueue")
    public void process2(Map message) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("[consumer-2]从队列 [CustomDirectQueue] 收到消息：" + message.toString());
    }

}

---

三、发送消息到队列：
下面只是调试样例，不涉及业务。

@RequestMapping("/producer")
@RestController
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("send")
    public String sendDirectMessage() {
        int i = 0;
        while(i < 100000) {
            HashMap map = new HashMap();
            map.put("message_id", UUID.randomUUID());
            map.put("message_msg", "这是一条消息数据");
            map.put("message_num", ++i);
            map.put("message_created_time", new Date());
            rabbitTemplate.convertAndSend("CustomDirectExchange", "custom.direct.routing", map);
        }
        return "发送成功";
    }

}

---

使用Postman进行模拟消息发送，最终可以看到两个消费者同时在处理这个队列的消息，默认使用的是轮询方式将消息分发给两个消费者。若某一个消费者处理的比较慢，依旧会导致应分发到该消费者的消息存在挤压情况。