RabbitMQ（三）：Java - 发布订阅模式

Java - 发布/订阅模式

发布/订阅模式(Publish/Subscribe)引入pom生产者消费者·接收短信消费者·接收邮件补充：结合工作队列模式

---

发布/订阅模式(Publish/Subscribe)

一个消息 ， 多个接收

特点：

一个生产者将消息发送给交换机与交换机绑定的有多个队列，每个消费者监听自己的队列生产者将消息发送给交换机，交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。如果消息发送给没有绑定队列的交换机上，消息将丢失。他比工作队列模式(work queue)更为强大，可以看成多个工作队列。

引入pom

<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> </dependency>

生产者

// 这里模拟注册完成后，发送短信和邮件。 public class Producer { // 定义两个队列和一个交换机 private final static String QUEUE_EMAIL = "queue_email"; private final static String QUEUE_SMS = "queue_sms"; private final static String EXCHANGE_NAME = "exchange_fanout_1"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = null; Channel channel = null; try{ connection = factory.newConnection(); // 创建内部会话通道，生产者和mq服务通信都在channel中完成 channel = connection.createChannel(); // 声明交换机 /** * 1. 交换机名称 * 2、交换机类型: * FANOUT:对应的模式就是 发布/订阅模式 * DIRECT:对应 路由(Routing) 的工作模式 * TOPIC:对应 Topics 工作模式 * HEADERS: 对应 HEADERS 工作模式 */ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 声明创建队列，如果没有，则创建 channel.queueDeclare(QUEUE_EMAIL,false,false,false, null); channel.queueDeclare(QUEUE_SMS,false,false,false,null); // 交换机队列绑定 /** * 1 queue 队列名称 * 2 exchange 交换机名称 * 3 routingKey 路由Key 发布订阅模式中 设置为"" */ channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,""); channel.queueBind(QUEUE_SMS,EXCHANGE_NAME,""); String message = "接好了，SMS和EMAIL~~"; // 发送消息 // 参数：String exchange, String routingKey, BasicProperties props, byte[] body /** * exchange:交换机 如果不指定("")，就默认交换机 * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称 * props:额外属性 * body：消息内容 */ channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes()); System.out.println("send message: "+message); } catch (TimeoutException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } finally { channel.close(); connection.close(); } } }

消费者·接收短信

/** * 短信 消费者 */ public class Consumer_SMS { private final static String QUEUE_EMAIL = "queue_email"; private final static String QUEUE_SMS = "queue_sms"; private final static String EXCHANGE_NAME = "exchange_fanout_1"; public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = null; Channel channel = null; try{ connection = factory.newConnection(); // 创建内部会话通道，生产者和mq服务通信都在channel中完成 channel = connection.createChannel(); // 声明交换机 /** * 1. 交换机名称 * 2、交换机类型: * FANOUT:对应的模式就是 发布/订阅模式 * DIRECT:对应路由的工作模式 * TOPIC:对应 Topics 工作模式 * HEADERS: 对应 HEADERS 工作模式 */ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 声明队列，如果没有，则创建 // channel.queueDeclare(QUEUE_EMAIL,false,false,false, null); channel.queueDeclare(QUEUE_SMS,false,false,false,null); // 交换机队列绑定 /** * 1 queue 队列名称 * 2 exchange 交换机名称 * 3 routingKey 路由Key 发布订阅模式中 设置为"" */ // channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,""); channel.queueBind(QUEUE_SMS,EXCHANGE_NAME,""); DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String (body, StandardCharsets.UTF_8); System.out.println("SMS receive:" + message); } }; // 接收消息 监听队列 // 参数：String queue, boolean autoAck, Consumer callback /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */ channel.basicConsume(QUEUE_SMS,true,consumer); } catch (TimeoutException | IOException e) { e.printStackTrace(); } } }

消费者·接收邮件

public class Consumer_EMAIL { private final static String QUEUE_EMAIL = "queue_email"; private final static String QUEUE_SMS = "queue_sms"; private final static String EXCHANGE_NAME = "exchange_fanout_1"; public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = null; Channel channel = null; try{ connection = factory.newConnection(); // 创建内部会话通道，生产者和mq服务通信都在channel中完成 channel = connection.createChannel(); // 声明交换机 /** * 1. 交换机名称 * 2、交换机类型: * FANOUT:对应的模式就是 发布/订阅模式 * DIRECT:对应路由的工作模式 * TOPIC:对应 Topics 工作模式 * HEADERS: 对应 HEADERS 工作模式 */ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 声明队列，如果没有，则创建 channel.queueDeclare(QUEUE_EMAIL,false,false,false, null); // 交换机队列绑定 /** * 1 queue 队列名称 * 2 exchange 交换机名称 * 3 routingKey 路由Key 发布订阅模式中 设置为"" */ channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,""); DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String (body, StandardCharsets.UTF_8); System.out.println("Email receive:" + message); } }; // 接收消息 监听队列 // 参数：String queue, boolean autoAck, Consumer callback /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */ channel.basicConsume(QUEUE_EMAIL,true,consumer); } catch (TimeoutException | IOException e) { e.printStackTrace(); } } }

补充：结合工作队列模式

其实只要在 消费者【c1】并列加上一个【c3】即可，c1、c3共享同一队列

// 新增消费者 Consumer_SMS_2 public class Consumer_SMS_2 { private final static String QUEUE_EMAIL = "queue_email"; private final static String QUEUE_SMS = "queue_sms"; private final static String EXCHANGE_NAME = "exchange_fanout_1"; public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = null; Channel channel = null; try{ connection = factory.newConnection(); // 创建内部会话通道，生产者和mq服务通信都在channel中完成 channel = connection.createChannel(); // 声明交换机 /** * 1. 交换机名称 * 2、交换机类型: * FANOUT:对应的模式就是 发布/订阅模式 * DIRECT:对应路由的工作模式 * TOPIC:对应 Topics 工作模式 * HEADERS: 对应 HEADERS 工作模式 */ channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 声明队列，如果没有，则创建 // channel.queueDeclare(QUEUE_EMAIL,false,false,false, null); channel.queueDeclare(QUEUE_SMS,false,false,false,null); // 交换机队列绑定 /** * 1 queue 队列名称 * 2 exchange 交换机名称 * 3 routingKey 路由Key 发布订阅模式中 设置为"" */ // channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,""); channel.queueBind(QUEUE_SMS,EXCHANGE_NAME,""); DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String (body, StandardCharsets.UTF_8); System.out.println("SMS_2 receive:" + message); } }; // 接收消息 监听队列 // 参数：String queue, boolean autoAck, Consumer callback /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */ channel.basicConsume(QUEUE_SMS,true,consumer); } catch (TimeoutException | IOException e) { e.printStackTrace(); } } }