介绍
我们都是基于一个队列发送和接受消息。
前面讲的几种，不管是生产者端还是消费者端都必须知道一个指定的QueueName才能发送、获取消息。 而RabbitMQ消息模型的核心思想是生产者不会将消息直接发送给队列。现在介绍一下完整的消息传递模式：
如果同一个消息，要求每个消费者都处理的话，就需要RabbitMQ提供的消息分发模式中的------"Publish/Subscribe"发布/订阅模式。
因为，生产者通常不会知道消息会被哪些消费者接收。
下面将的模式，生产者虽然不是直接将消息发送给Queue，但是会交给Exchange，所以需要定义Exchange的消息分发模式 ，之前的程序中，有如下一行代码：
channel.basicPublish("", queueName , null , msg.getBytes());
第一个参数为空字符串，其实第一个参数就是用来指定转发器名称----ExchangeName的，这里用空字符串，就表示消息会交给默认的Exchange。
Exchange----转发器
它的作用是一方面它接受生产者的消息，另一方面向队列推送消息。
转发器必须清楚的知道如何处理接收到的消息。附加一个特定的队列吗？附加多个队列？或者是否丢弃？这些规则通过转发器的类型进行定义。
Exchange的类型有：Direct、Topic（主题模式）、Headers和Fanout。
1.Fanout--广播模式
我们可以通过以下代码创建一个指定类型的转发器：
[java] view plain copy channel.exchangeDeclare("logs", "fanout"); fanout转发器非常简单，从名字就可以看出，它是广播接受到的消息给所有的队列。 现在我们可以指定转发器的名字了：
[java] view plain copy channel.basicPublish( "logs", "", null, message.getBytes());
你可能还记得之前我们用队列时，会指定一个名字。队列有名字对我们来说是非常重要的——我们需要为消费者指定同一个队列。
但这并不是我们的日志系统所关心的。我们要监听所有日志消息，而不仅仅是一类日志。我们只对对当前流动的消息感兴趣。解决这些问题，我们需要完成两件事。
首先，每当我盟连接到RabbitMQ时，需要一个新的空队列。为此我们需要创建一个随机名字的空队列，或者更好的，让服务器选好年则一个随机名字的空队列给我们。
其次，一旦消费者断开连接，队列将自动删除。
我们提供一个无参的queueDeclare()方法，创建一个非持久化、独立的、自动删除的队列，且名字是随机生成的。
[java] view plain copy String queueName = channel.queueDeclare().getQueue();
Binding----绑定
我们创建了一个广播的转发器和随机队列。我们需要告诉转发器转发消息的队列。这个关联转发器和队列的我们叫它Binding。[java] view plain copy channel.queueBind(queueName, "logs", ""); 完整代码： 发送端：
[java] view plain copy 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(); } } 接收端--（两个一样）： [java] view plain copy 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---直接指定模式
任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。
1.一般情况可以使用rabbitMQ自带的Exchange：”"(该Exchange的名字为空字符串，下文称其为default Exchange)。
2.这种模式下不需要将Exchange进行任何绑定(binding)操作
3.消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。
4.如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。
生产者端代码：
[java] view plain copy 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与生产者相同
[java] view plain copy 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与生产者不同 [java] view plain copy 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 + "'"); } } } 注意： 启动时先启动一次生产者，然后再启动消费者。如果没有接收到消息再启动一次生产者即可达到想要的效果。
