RabbitMQ消息分发模式----"Publish/Subscribe"发布/订阅模式

深藏阁楼爱情的钟 2022-08-20 04:18 240阅读 0赞

**介绍**

我们都是基于一个队列发送和接受消息。

前面讲的几种，不管是生产者端还是消费者端都必须知道一个指定的QueueName才能发送、获取消息。  而RabbitMQ消息模型的核心思想是生产者不会将消息直接发送给队列。现在介绍一下完整的消息传递模式：

如果同一个消息，要求每个消费者都处理的话，就需要RabbitMQ提供的消息分发模式中的------"Publish/Subscribe"发布/订阅模式。

因为，生产者通常不会知道消息会被哪些消费者接收。

下面将的模式，生产者虽然不是直接将消息发送给Queue，但是会交给Exchange，所以需要定义Exchange的消息分发模式 ，之前的程序中，有如下一行代码：

channel.basicPublish("", queueName , null , msg.getBytes());

第一个参数为空字符串，其实第一个参数就是用来指定转发器名称----ExchangeName的，这里用空字符串，就表示消息会交给默认的Exchange。

**Exchange----转发器**

它的作用是一方面它接受生产者的消息，另一方面向队列推送消息。

转发器必须清楚的知道如何处理接收到的消息。附加一个特定的队列吗？附加多个队列？或者是否丢弃？这些规则通过转发器的类型进行定义。

Exchange的类型有：Direct、Topic（主题模式）、Headers和Fanout。

1.Fanout--广播模式

![Center][]

我们可以通过以下代码创建一个指定类型的转发器：

channel.exchangeDeclare("logs", "fanout");

fanout转发器非常简单，从名字就可以看出，它是广播接受到的消息给所有的队列。  
现在我们可以指定转发器的名字了：

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

你可能还记得之前我们用队列时，会指定一个名字。队列有名字对我们来说是非常重要的——我们需要为消费者指定同一个队列。

但这并不是我们的日志系统所关心的。我们要监听所有日志消息，而不仅仅是一类日志。我们只对对当前流动的消息感兴趣。解决这些问题，我们需要完成两件事。

首先，每当我盟连接到RabbitMQ时，需要一个新的空队列。为此我们需要创建一个随机名字的空队列，或者更好的，让服务器选好年则一个随机名字的空队列给我们。

其次，一旦消费者断开连接，队列将自动删除。

我们提供一个无参的queueDeclare()方法，创建一个非持久化、独立的、自动删除的队列，且名字是随机生成的。

String queueName = channel.queueDeclare().getQueue();

**Binding----绑定**

我们创建了一个广播的转发器和随机队列。我们需要告诉转发器转发消息的队列。这个关联转发器和队列的我们叫它Binding。

![Center 1][]

channel.queueBind(queueName, "logs", "");

**完整代码：**  
**发送端：**

package cn.rabbitmq.publish;
    import java.io.IOException;
    import cn.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class SendTest {
        private final static String EXCHANGE_NAME = "logs";  
        public static void main(String[] args) throws IOException {  
            /** 
             * 创建连接连接到MabbitMQ 
             */  
            Connection connection = ConnectionUtil.getConnection();
            // 创建一个频道  
            Channel channel = connection.createChannel();  
            // 指定转发——广播  
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");  
              for(int i=0;i<3;i++){  
                // 发送的消息  
                String message = "Hello World!";  
                channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());  
                System.out.println(" [x] Sent '" + message + "'");  
            }          
            // 关闭频道和连接  
            channel.close();  
            connection.close();  
        }  
    }

**接收端--（两个一样）：**

package cn.rabbitmq.publish;
    import java.io.IOException;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    import cn.rabbitmq.util.ConnectionUtil;
    public class FanoutRecv1 {
        private static final String EXCHANGE_NAME = "logs";
        public static void main(String[] argv) throws IOException, InterruptedException {
            Connection connection = ConnectionUtil.getConnection();
            final Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            // 声明一个随机队列
            String queueName = channel.queueDeclare().getQueue();
            channel.queueBind(queueName, EXCHANGE_NAME, "");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 指定接收者，第二个参数为自动应答，无需手动应答
            channel.basicConsume(queueName, true, consumer);
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
            }
        }
    }

2.driect---直接指定模式

![Center 2][]

任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。

1.一般情况可以使用rabbitMQ自带的Exchange：”"(该Exchange的名字为空字符串，下文称其为default Exchange)。

2.这种模式下不需要将Exchange进行任何绑定(binding)操作

3.消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。

4.如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

**生产者端代码：**

package cn.rabbitmq.publish;
    import java.io.IOException;
    import cn.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    public class SendTest {
        private final static String EXCHANGE_NAME = "logs1";  
            public static void main(String[] args) throws IOException {  
            /** 
             * 创建连接连接到MabbitMQ 
             */  
            Connection connection = ConnectionUtil.getConnection();
            // 创建一个频道  
            Channel channel = connection.createChannel();  
            // 指定转发——广播  
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");  
      
            for(int i=0;i<3;i++){  
                // 发送的消息  
                String message = "Hello World!";  
                channel.basicPublish(EXCHANGE_NAME, "key", null, message.getBytes());  
                System.out.println(" [x] Sent '" + message + "'");  
            }  
            // 关闭频道和连接  
            channel.close();  
            connection.close();  
        }  
    }

**消费者端代码：**

**消费者一：routKey与生产者相同**

package cn.rabbitmq.publish;
    import java.io.IOException;
    import cn.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    public class FanoutRecv2 {
        private static final String EXCHANGE_NAME = "logs1";
        private final static String QUEUE_NAME = "logs1";
        public static void main(String[] argv) throws IOException, InterruptedException {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            // 声明一个随机队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 指定接收者，第二个参数为自动应答，无需手动应答
            channel.basicConsume(QUEUE_NAME, true, consumer);
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
            }
        }
    }<strong>
    </strong>

**消费者二：routKey与生产者不同**

package cn.rabbitmq.publish;
    import java.io.IOException;
    import cn.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    public class FanoutRecv2 {
        private static final String EXCHANGE_NAME = "logs1";
        private final static String QUEUE_NAME = "logs1";
        public static void main(String[] argv) throws IOException, InterruptedException {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            // 声明一个随机队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key1");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 指定接收者，第二个参数为自动应答，无需手动应答
            channel.basicConsume(QUEUE_NAME, true, consumer);
    <span style="white-space:pre">	</span>while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
            }
        }
    }

注意：启动时先启动一次生产者，然后再启动消费者。如果没有接收到消息再启动一次生产者即可达到想要的效果。

[http://blog.csdn.net/lmj623565791/article/details/37657225][http_blog.csdn.net_lmj623565791_article_details_37657225]

[Center]: /images/20220731/d901b7a94ac64e60a39db766e14ebd08.png
[Center 1]: /images/20220731/9d05216edbb84d5aa8f45d1afaaa8b63.png
[Center 2]: /images/20220731/f3033ff0c7914ac986e8d6146a48b111.png
[http_blog.csdn.net_lmj623565791_article_details_37657225]: http://blog.csdn.net/lmj623565791/article/details/37657225