概述
fanout扇出也称之为广播
在广播模式下，消息发送的流程是这样的，如下所示：
- 可以有多个消费者。
- 每个消费者有自己的 queue（队列）
- 每个队列都要绑定到 Exchange（交换机）
- 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
- 交换机把消息发送给绑定过的所有队列。
- 队列的消费者都能拿到消息。实现一条消息被多个消费者消费。
创建生产者
/**
* @author: BNTang
*/
public class Producer {
@Test
public void sendMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
// 创建通道
Channel channel = connection.createChannel();
// 设置交换机
channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
// 向交换机发消息
channel.basicPublish("logs", "", null, ("我是个 fanout 类型的消息").getBytes());
RabbitMQUtil.closeChannelAndConnection(channel, connection);
System.out.println("消息发送成功");
}
}
创建消费者 1
/**
* @author BNTang
*/
public class Consumer1 {
@Test
public void receiveMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
// 得到通道
Channel channel = connection.createChannel();
// 绑定交换机
channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
// 从通道里面得到一个临时的队列
String queue = channel.queueDeclare().getQueue();
// 把临时队列和交换机进行绑定
channel.queueBind(queue, "logs", "");
// 接收消息
channel.basicConsume(queue, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者【1】接收到消息" + new String(body));
}
});
System.out.println("消费者【1】启动成功");
System.in.read();
}
}
创建消费者 2
/**
* @author BNTang
*/
public class Consumer2 {
@Test
public void receiveMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
// 得到通道
Channel channel = connection.createChannel();
// 绑定交换机
channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
// 从通道里面得到一个临时队列
String queue = channel.queueDeclare().getQueue();
// 把临时的队列和交换机进行绑定
channel.queueBind(queue, "logs", "");
// 接收消息
channel.basicConsume(queue, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者【2】接收到消息" + new String(body));
}
});
System.out.println("消费者【2】启动成功");
System.in.read();
}
}
测试方式，先启动消费者1，和消费者2，在启动消息生产者即可。
