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

Java - 发布/订阅模式

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

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

一个消息，多个接收

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

特点：

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

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

Java - 发布/订阅模式

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

引入pom

生产者

消费者·接收短信

消费者·接收邮件

补充：结合工作队列模式

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method