RabbitMQ示例生产者消费者（二）——Exchange fanout方式

你的名字 2022-06-10 13:20 193阅读 0赞

# RabbitMQ示例生产者消费者（二）——Exchange fanout方式 #

本文主要参考：

http://www.rabbitmq.com/tutorials/tutorial-three-java.html

好，先介绍一下本篇文章中的主要内容。

本文主要介绍的是通过fanout的exchange方式，进行生产，消费端通过binding操作，对queue和exchange进行绑定，然后消费。

好，下面看一下代码：

生产者：

package com.xueyou.demo;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /**
     * Created by wuxueyou on 2017/8/20.
     */
    public class AppPublish {
        private final static String EXCHANGE_NAME = "log-exchange";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.0.99");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
    
    
            int i = 10000;
            while (i-- > 0) {
                String message = "hey this is the loginfo " + i;
                channel.basicPublish(EXCHANGE_NAME,"", null, message.getBytes());
                System.out.println("sent message:" + message);
                Thread.sleep(100);
            }
            channel.close();
            connection.close();
        }
    }

消费者：

package com.xueyou.demo;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    /**
     * Created by wuxueyou on 2017/8/20.
     */
    public class AppSubScribe {
        private final static String EXCHANGE_NAME = "log-exchange";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.0.99");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            String queueName = channel.queueDeclare().getQueue();
            channel.queueBind(queueName, EXCHANGE_NAME, "");
    
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("receive message :" + message);
                }
            };
            channel.basicConsume(queueName, true, consumer);
        }
    }

在生产者和消费者端需要注意的是：生产者端使用的是fanout的方式，所以在进行发送的时候只需要进行exchange的声明，而不需要对routingkey进行指定；而消费者端是这样，消费者端需要声明一个queue，同时把queue和exchange进行绑定，这样才能进行消费。

fanout方式是这样，是一种广播的方式，只要和exchange绑定的queue都能够接受到exchange发送的消息，而且每个queue中的内容都是全量的。