发布订阅模型 Publish/Subscribe
发布订阅模型也称为广播模型，交换机类型需要指定为Fanout，正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的所有队列中。每个消费者都监听自己的队列，所以同一个消息，会被所有的消费者共同消费。Fanout 这种交换类型并不能给我们带来很大的灵活性，它只能进行无意识的广播。
在广播模式下，消息发送流程是这样的：
- 可以有多个消费者。
- 每个消费者有自己的Queue。
- 每个队列都要绑定到Exchange。
- 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
- 交换机把消息发送给绑定过的所有队列。
- 队列的消费者都能拿到消息。实现一条消息被多个消费者消费。
创建生产者
public class MyProducer { @Test public void test() throws Exception { // Fanout模式不需要指定队列 String queue = ""; // 交换机 String exchange = "logs"; // 创建工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setVirtualHost("/"); factory.setHost("xuewei.world"); factory.setUsername("xuewei"); factory.setPassword("123456"); factory.setPort(5672); // 创建连接和通道 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明交换机 channel.exchangeDeclare(exchange, "fanout"); for (int i = 0; i < 3; i++) { // 发布消息 channel.basicPublish(exchange, queue, null, ("DEBUG LOG -> " + i).getBytes()); channel.basicPublish(exchange, queue, null, ("INFO LOG -> " + i).getBytes()); channel.basicPublish(exchange, queue, null, ("WARN LOG -> " + i).getBytes()); channel.basicPublish(exchange, queue, null, ("ERROR LOG -> " + i).getBytes()); } } }
创建消费者1
public class MyConsumer1 { public static void main(String[] args) throws Exception { // 指定交换机 String exchange = "logs"; // 创建工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setVirtualHost("/"); factory.setHost("xuewei.world"); factory.setUsername("xuewei"); factory.setPassword("123456"); factory.setPort(5672); // 创建连接和通道 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); //绑定交换机 channel.exchangeDeclare("logs", "fanout"); //创建临时队列 String queue = channel.queueDeclare().getQueue(); //将临时队列绑定exchange channel.queueBind(queue, "logs", ""); //处理消息 channel.basicConsume(queue, true, 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)); // TODO 业务处理 } }); } }
创建消费者2
public class MyConsumer2 { public static void main(String[] args) throws Exception { // 指定交换机 String exchange = "logs"; // 创建工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setVirtualHost("/"); factory.setHost("xuewei.world"); factory.setUsername("xuewei"); factory.setPassword("123456"); factory.setPort(5672); // 创建连接和通道 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); //绑定交换机 channel.exchangeDeclare("logs", "fanout"); //创建临时队列 String queue = channel.queueDeclare().getQueue(); //将临时队列绑定exchange channel.queueBind(queue, "logs", ""); //处理消息 channel.basicConsume(queue, true, 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)); } }); } }
两个消费者同时都收到了消息。
