RabbitMQ 发布订阅模式(Publish、Subscribe)

private final static String EXCHANGE_NAME = "test_exchange_fanout";
@Test
public void testSend() throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明exchange
    channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    // 消息内容
    String message = "订阅消息";
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");
    channel.close();
    connection.close();
}

注意：消息发送到没有队列绑定的交换机时，消息将丢失，因为交换机没有存储消息的能力，消息只能存在在队列中。

private final static String QUEUE_NAME_1 = "test_queue_exchange_1";
private final static String QUEUE_NAME_2 = "test_queue_exchange_2";
@Test
public void testRecv1() throws Exception {
    recv(QUEUE_NAME_1,EXCHANGE_NAME);
}
@Test
public void testRecv2() throws Exception {
    recv(QUEUE_NAME_2,EXCHANGE_NAME);
}
public void recv(String queueName,String exchangeName) throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(queueName, false, false, false, null);
    // 绑定队列到交换机
    channel.queueBind(queueName, exchangeName, "");
    // 同一时刻服务器只会发一条消息给消费者
    channel.basicQos(1);
    // 定义队列的消费者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 监听队列，手动返回完成
    channel.basicConsume(queueName, false, consumer);
    // 获取消息
    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        String message = new String(delivery.getBody());
        System.out.println(" [x] Received '" + message + "'");
        Thread.sleep(10);
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}