Native RabbitMQ Fanout Exchange

RabbitMQ原生编程 ，Fanout广播交换器。广播交换器发送消息到队列，跟路由键没什么关系了，只要绑定到广播交换器的队列都会受到它发来的消息。

生产者：

/**
 * @author zhangjianbing
 * time 2020/9/3
 */
@SuppressWarnings("Duplicates")
public class FanoutProducer {

    public final static String EXCHANGE_NAME = "fanout_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {

        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("1.1.1.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("beijing");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("beijing");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 在信道中设置交换器，这里将第三个参数设置成true，交换器持久化
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT, true);

        // 声明一个普通的队列
        String queueName1 = "EAT.CALLBACK.QUEUE";
        // 这里第二个参数设置成true，持久队列
        channel.queueDeclare(queueName1, true, false, false, null);

        // 声明一个普通的队列
        String queueName2 = "SLEEP.CALLBACK.QUEUE";
        // 这里第二个参数设置成true，持久队列
        channel.queueDeclare(queueName2, true, false, false, null);

        // 设置路由键
        String routeKey = "fanout_routeKey";
        // 将队列和交换器通过相同的路由键绑定在一起
        channel.queueBind(queueName1, EXCHANGE_NAME, routeKey);
        channel.queueBind(queueName2, EXCHANGE_NAME, routeKey);

        // 消息
        String message = "hello rabbit message queue";
        // 发送消息
        channel.basicPublish(EXCHANGE_NAME, routeKey, null, message.getBytes());
        System.out.println("消息发送成功。。。。。。");

        // 关闭信道和连接
        channel.close();
        connection.close();

    }

}

消费者：

/**
 * @author zhangjianbing
 * time 2020/9/3
 */
@SuppressWarnings("Duplicates")
public class FanoutConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("1.1.1.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("beijing");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("beijing");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();

        System.out.println("正在等待消息。。。。。。");

        // 声明一个消费者
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.out.println("【路由键】：" + envelope.getRoutingKey() + "【消息内容】：" + new String(body, StandardCharsets.UTF_8));
            }
        };

        // 指定队列去消费
        String queueName1 = "EAT.CALLBACK.QUEUE";
        // 消费者正式开始在指定队列上消费[队列名称、自动提交、消费者]
        channel.basicConsume(queueName1, true, consumer);

        // 指定队列去消费
        String queueName2 = "SLEEP.CALLBACK.QUEUE";
        // 消费者正式开始在指定队列上消费[队列名称、自动提交、消费者]
        channel.basicConsume(queueName2, true, consumer);

    }

}