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

红太狼 2021-09-24 15:30 362阅读 0赞

### Java - 发布/订阅模式 ###

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

--------------------

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

> 一个消息 ， 多个接收

![发布订阅者模式][20200903095438890.png_pic_center]

特点：

*  一个生产者将消息发送给**交换机**
 *  与交换机绑定的有多个队列，每个消费者监听自己的队列
 *  生产者将消息发送给交换机，交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。
 *  如果消息发送给没有绑定队列的交换机上，消息将丢失。
 *  他比工作队列模式(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();
            }
        }
    }

[20200903095438890.png_pic_center]: /images/20210923/14e5dd948f054ade93010bc9c35560d5.png