在上一篇文章中介绍了rabbitmq的fanout模式。今天继续学习另一种模式——direct模式。这种模式是rabbitmq的最简单一种模式。

首先创建一个名为helloDirect1的对列

@Configuration
public class DirectRabbitConfig {

    @Bean
    public Queue directA(){
        return  new Queue("helloDirect1");
    }
}

创建发送端

@Component
public class DirectSend {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String context){
        System.out.println("Sender:" + context);
        // 这里需要指定发送的消息队列
        this.rabbitTemplate.convertAndSend("helloDirect1",context);
    }
}

接受端

@Component
@RabbitListener(queues = "helloDirect1")
public class DirectReceiveA {

    @RabbitHandler
    public void receive(String context){
        System.out.println("direct.A"+context);
    }
}

在这里，我有点不是特别清楚的地方，就是rabbitmq中说的是direct模式不需要exchange，只需要绑定routingKey即可，即

this.rabbitTemplate.convertAndSend("helloDirect1",context);中第一个参数为routingKey，但是我在写代码过程发现其实这块应该写的是你想发送到哪个对列，就写这个对列名称。不知道这样理解对不对，如果有不对的地方请指出。

控制台打印：

说明我们接受成功。

在写代码过程中突然想到，@RabbitListener注解是不是可以同时监听到多个对列呢？

在创建一个为ceshi的对列

@Bean
    public Queue directB(){
        return  new Queue("ceshi");
    }

同时监听多个对列

@Component
@RabbitListener(queues = {"helloDirect1","ceshi"})
public class DirectReceiveA {

    @RabbitHandler
    public void receive(String context){
        System.out.println("direct.A"+context);
    }
}

发送端

@Component
public class DirectSend {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String context){
        System.out.println("Sender:" + context);
        // 这里需要指定发送的消息队列
        this.rabbitTemplate.convertAndSend("ceshi",context);
        this.rabbitTemplate.convertAndSend("helloDirect1",context);
    }
}

控制台打印

成功

当生产者和消费者是一对多时，会是什么样的效果呢？

在创建一个消费端同时监听helloDirect1对列

@Component
@RabbitListener(queues = "helloDirect1")
public class DirectReceiveB {

    @RabbitHandler
    public void receive(String context){
        System.out.println("direct.B"+context);
    }
}

改一下测试类，循环发送10遍，看效果

@Test
	public void DirectSend(){
		for (int i=0;i<10;i++){
			directSend.send("测试"+i);
		}
	}

但是如果一个消费者消费快，另一个慢的话回怎么样呢？

我们对消费者B休眠2秒，看效果

@Component
@RabbitListener(queues = "helloDirect1")
public class DirectReceiveB {

    @RabbitHandler
    public void receive(String context){
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("direct.B"+context);
    }
}

会发现A消费者消费结束了，但B消费者还没有完成。这就造成了A消费者空闲下来了，资源浪费。