@Configuration
public class RabbitConfig {
    /**
     * 声明 Direct 交换机 支持持久化.
     *
     * @return the exchange
     */
    @Bean("directExchange")
    public Exchange directExchange() {
        return ExchangeBuilder.directExchange("DIRECT_EXCHANGE").durable(true).build();
    }

    /**
     * 声明 fanout 交换机.
     *
     * @return the exchange
     */
    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        return (FanoutExchange) ExchangeBuilder.fanoutExchange("FANOUT_EXCHANGE").durable(true).build();
    }

    /**
     * 声明一个队列 支持持久化.
     *
     * @return the queue
     */
    @Bean("queueA")
    public Queue directQueue() {
        return QueueBuilder.durable("QUEUE_A").build();
    }

    @Bean("queueB")
    public Queue directQueue2() {
        return QueueBuilder.durable("QUEUE_B").build();
    }

    @Bean("queueC")
    public Queue directQueue3() {
        return QueueBuilder.durable("QUEUE_C").build();
    }

    /**
     * 绑定队列A 到 direct 交换机.
     *
     * @param queue          the queue
     * @param directExchange the  exchange
     * @return the binding
     */
    @Bean
    public Binding bindingA(@Qualifier("queueA") Queue queue,
                            @Qualifier("directExchange") Exchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with("DIRECT_ROUTING_KEY_A").noargs();
    }

    @Bean
    public Binding bindingB(@Qualifier("queueB") Queue queue,
                            @Qualifier("directExchange") Exchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with("DIRECT_ROUTING_KEY_B").noargs();
    }

    /**
     * 绑定队列A 到 fanout 交换机.
     *
     * @param queue          the queue
     * @param fanoutExchange the  exchange
     * @return the binding
     */
    @Bean
    public Binding bindingA1(@Qualifier("queueA") Queue queue,
                             @Qualifier("fanoutExchange") Exchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange).with("FANOUT_ROUTING_KEY_A").noargs();
    }

    @Bean
    public Binding bindingA2(@Qualifier("queueB") Queue queue,
                             @Qualifier("fanoutExchange") Exchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange).with("FANOUT_ROUTING_KEY_B").noargs();
    }

    @Bean
    public Binding bindingA3(@Qualifier("queueC") Queue queue,
                             @Qualifier("fanoutExchange") Exchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange).with("FANOUT_ROUTING_KEY_C").noargs();
    }

}

@Service
public class SenderService {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 测试广播模式.
     *
     * @param p the p
     * @return the response entity
     */
    public void broadcast(String p) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("FANOUT_EXCHANGE", "", p, correlationData);
    }

    /**
     * 测试 Direct 模式.
     *
     * @param p the p
     * @return the response entity
     */
    public void directA(String p) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("DIRECT_EXCHANGE", "DIRECT_ROUTING_KEY_A", p, correlationData);
    }
    public void directB(String p) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("DIRECT_EXCHANGE", "DIRECT_ROUTING_KEY_B", p, correlationData);
    }
    public void directNull(String p) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("DIRECT_EXCHANGE", "", p, correlationData);
    }

}

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

    /**
     * FANOUT广播队列监听一.
     *
     * @param message the message
     * @param channel the channel
     * @throws IOException the io exception  这里异常需要处理
     */
    @RabbitListener(queues = {"QUEUE_A"})
    public void on(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("FANOUT_QUEUE_A " + new String(message.getBody()));
    }
    @RabbitListener(queues = {"QUEUE_B"})
    public void t(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("FANOUT_QUEUE_B " + new String(message.getBody()));
    }
    @RabbitListener(queues = {"QUEUE_C"})
    public void t(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("FANOUT_QUEUE_C " + new String(message.getBody()));
    }

    /**
     * DIRECT模式.
     *
     * @param message the message
     * @param channel the channel
     * @throws IOException the io exception  这里异常需要处理
     */
    @RabbitListener(queues = {"DIRECT_QUEUE"})
    public void message(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("DIRECT_QUEUE " + new String(message.getBody()));
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class SenderServiceTest {
    @Autowired
    private SenderService senderService;

    @Test
    public void testCache() throws InterruptedException {
        // 测试广播模式
        senderService.broadcast("   Test    同学们集合啦！");
        // 测试Direct模式
        senderService.directA("   Test    定点消息 A ");
        senderService.directB("   Test    定点消息 B ");
        senderService.directNull("   Test    定点消息 null key ");
        Thread.sleep(5000L);
    }
}