RabbitMQ消息分发模式----"Publish/Subscribe"发布/订阅模式-蒲公英云

介绍

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

前面讲的几种，不管是生产者端还是消费者端都必须知道一个指定的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 + "'");
        }
    }
}