RabbitMQ-交换机模式

在说正题之前先解释一下交换机模式是个笼统的称呼，它不是一个单独的模式（包括了订阅模式，路由模式和主题模式），交换机模式是一个比较常用的模式，主要是为了实现数据的同步。

首先，说一下订阅模式，就和字面上的意思差不多主要就是一个生产者，多个消费者，同一个消息被多个消费者获取，先看一下官网的图示

整体执行过程就和图里一样，生产者把消息发送到交换机，然后队列绑定到交换机，消息由交换机发送到队列，每一个队列都有一个各自的消费者。这样

就实现了一个消息被多个消费者所获取，而且如果有新的消费者加入直接绑定队列到交换机就可以了，大大的降低了系统间的耦合度。还有一点要注意的就是

当我们把消息发送到一个没有队列绑定的交换机时，消息就会丢失，因为消息只能存储在队列，而交换机只做交换，不做存储！

生产者代码：

public class Send {

    private final static String EXCHANGE_NAME = "exchange_name"; //交换机名称

    public static void main(String[] argv) throws Exception {

        // 获取MQ连接和通道

        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        // 声明交换机

        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 消息内容

        String message = "生产者消息";

        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());

        System.out.println(" 发送 '" + message + "'");

        channel.close();

        connection.close();

    }

}

消费者一号代码：

public class Recv {

    private final static String QUEUE_NAME = "test_queue_ex";

    private final static String EXCHANGE_NAME = "exchange_name";

    public static void main(String[] argv) throws Exception {

        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        // 声明队列

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机

        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        channel.basicQos(1);

        // 定义队列的消费者

        QueueingConsumer consumer = new QueueingConsumer(channel);

        // 监听队列，手动返回完成

        channel.basicConsume(QUEUE_NAME, false, consumer);

        // 获取消息

        while (true) {

            QueueingConsumer.Delivery delivery = consumer.nextDelivery();

            String message = new String(delivery.getBody());

            System.out.println(" 消费者一号 '" + message + "'");

            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

        }

    }

}

消费者二号代码：

public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_ex2";

    private final static String EXCHANGE_NAME = "exchange_name";

    public static void main(String[] argv) throws Exception {

        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);

        channel.basicConsume(QUEUE_NAME, false, consumer);

        while (true) {

            QueueingConsumer.Delivery delivery = consumer.nextDelivery();

            String message = new String(delivery.getBody());

            System.out.println("消费者二号 '" + message + "'");

            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

        }

    }

}