RabbitMQ——使用Exchange中的fanout交换机实现消息发送和接收

Posted 2021-06-30 张起灵-小哥

文章目录：

1.写在前面

2.使用fanout交换机实现消息的发送和接收

2.1 编写消息接收类（有两个）

2.2 编写消息发送类

---

1.写在前面

所有 MQ 产品从模型抽象上来说都是一样的过程：
消费者（consumer）订阅某个队列。生产者（producer）创建消息，然后发布到队列（queue）中，最后将消息发送到监听的消费者。

上面是MQ的基本抽象模型，但是不同的MQ产品有有者不同的机制，RabbitMQ实际基于AMQP协议的一个开源实现，因此RabbitMQ内部也是AMQP的基本概念。

RabbitMQ的内部接收如下：

1、Message
消息，消息是不具体的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。

2、Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。

3、Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

4、Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

5、Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

6、Connection
网络连接，比如一个TCP连接。

7、Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

8、Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

9、Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。

10、Broker
表示消息队列服务器实体。

---

2.使用fanout交换机实现消息的发送和接收

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

注意：fanout模式的消息需要将一个消息同时绑定到多个队列中因此这里不能创建并指定某个队列。

注意：

1、使用fanout模式获取消息时不需要绑定特定的队列名称，只需使用channel.queueDeclare().getQueue();获取一个随机的队列名称，然后绑定到指定的Exchange即可获取消息。

2、这种模式中可以同时启动多个接收者只要都绑定到同一个Exchang即可让所有接收者同时接收同一个消息是一种广播的消息机制

Fanout交换机中是一种广播模式，消息是一对多的。这种模式种，没有RoutingKey以及BindingKey的概念，Bindings只是简单的将消息与交换机进行了绑定，如果消息进入了交换机中，那么这个消息会被转发到所有与当前交换机进行绑定的所有队列中。这种模式就像我们收看电视或者电台直播一样，必须要先打开消息接收者来监听队列（就像要先打开电视等待节目开始），这个时候只要有消息发送过来，那么所有的监听者都可以收到消息；如果没有提前监听队列，那么一旦消息发送了，消息接收者就可能错过这条消息。

也就是Fanout交换机模式下，它是会丢失数据的，但是它的速度是最快的。

---

2.1 编写消息接收类（有两个）

package com.szh.rabbitmq.exchange.fanout;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Receive01 {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            /**
             * 由于Fanout类型的交换机的消息模式类似于广播模式，它不需要绑定RoutingKey
             * 又有可能会有很多个消费者来接收这个交换机中的数据，因此我们创建队列时，要创建一个随机的队列名称
             *
             * queueDeclare()方法会创建一个随机名称的一个队列，非持久化，排外的（最多允许一个消费者监听该队列）
             * 同时也是自动删除的，当没有消费者监听这个队列时，它会自动删除
             *
             * getQueue()方法用于获取这个随机队列的名称
             */
            String queueName=channel.queueDeclare().getQueue();
            channel.exchangeDeclare("fanoutExchange","fanout",true);
            channel.queueBind(queueName,"fanoutExchange","");
            //监听某个消息队列，同时获取消息队列中的数据
            channel.basicConsume(queueName,true,"",new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message=new String(body);
                    System.out.println("Receive01-消息接收成功：" + message);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

package com.szh.rabbitmq.exchange.fanout;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Receive02 {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            /**
             * 由于Fanout类型的交换机的消息模式类似于广播模式，它不需要绑定RoutingKey
             * 又有可能会有很多个消费者来接收这个交换机中的数据，因此我们创建队列时，要创建一个随机的队列名称
             *
             * queueDeclare()方法会创建一个随机名称的一个队列，非持久化，排外的（最多允许一个消费者监听该队列）
             * 同时也是自动删除的，当没有消费者监听这个队列时，它会自动删除
             *
             * getQueue()方法用于获取这个随机队列的名称
             */
            String queueName=channel.queueDeclare().getQueue();
            channel.exchangeDeclare("fanoutExchange","fanout",true);
            channel.queueBind(queueName,"fanoutExchange","");
            //监听某个消息队列，同时获取消息队列中的数据
            channel.basicConsume(queueName,true,"",new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message=new String(body);
                    System.out.println("Receive02-消息接收成功：" + message);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

此时直接运行这两个消息接收者，使得它们俩一直处于对消息队列的监听状态，一旦有消息发送，则会立刻接收到消息。

运行之后，在RabbitMQ的管理界面可以看到Queues中，会生成两个随机名称的消息队列。

2.2 编写消息发送类

package com.szh.rabbitmq.exchange.fanout;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 *
 */
public class Send {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection=null;
        Channel channel=null;

        try {
            connection=factory.newConnection();
            channel=connection.createChannel();

            /**
             * 由于使用Fanout类型的交换机，因此消息的接收方可能会有多个，因此不建议在消息发送时，创建队列
             * 同时也不建议将该队列绑定到fanout交换机中，因为一旦绑定了一个队列，那么其他队列将无法获得消息
             * 但是发送消息时，至少应该确保交换机存在
             */
//            channel.queueDeclare("myDirectQueue",true,false,false,null);
            channel.exchangeDeclare("fanoutExchange","fanout",true);
//            channel.queueBind("myDirectQueue","directExchange","directRoutingKey");

            String message="Exchange的fanout消息绑定";
            channel.basicPublish("fanoutExchange","",null,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("消息发送成功：" + message);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null ) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

运行消息发送者即可，可以看到两个消息接收者都正确无误的接收到了数据。