3.3 发布订阅模式

在开发过程中，有一些消息需要不同消费者进行不同的处理，如电商网站的同一条促销信息需要短信发送、邮件发送、站内信发送等。此时可以使用发布订阅模式(Publish/Subscribe)

特点

生产者将消息发送给交换机，交换机将消息转发到绑定此交换机的每个队列中。
工作队列模式的交换机只能将消息发送给一个队列，发布订阅模式的交换机能将消息发送给多个队列。发布订阅模式使用fanout交换机。

1、编写生产者

package com.zj.mq.publish;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*生产者*/
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("MQzhang");
        connectionFactory.setPassword("MQzhang");
        connectionFactory.setVirtualHost("/");
        //2.创建连接
        Connection connection = connectionFactory.newConnection();
        //3.建立信道
        Channel channel = connection.createChannel();
        //4.创建交换机fanout
        /*
        * 参数一：交换机名称
        * 参数二：交换机类型
        * 参数三： 交换机是否持久化（关闭控制台是否还存在）*/
        channel.exchangeDeclare("exchangeFanout", BuiltinExchangeType.FANOUT,false);
        //5.创建三个队列(分别模拟邮件发送、短信发送、站内信发送)
        channel.queueDeclare("mailQueue", false,false,false,null);
        channel.queueDeclare("messageQueue", false,false,false,null);
        channel.queueDeclare("stationQueue", false,false,false,null);
        //6.交换机绑定队列
        /*
        * 参数一：队列名称
        * 参数二：交换机名称
        * 参数三：路由关键字，发布订阅模式不存在路由关键字*/
        channel.queueBind("mailQueue","exchangeFanout","");
        channel.queueBind("messageQueue","exchangeFanout","");
        channel.queueBind("stationQueue","exchangeFanout","");
        //7.往交换机发送消息
        for (int i = 0; i < 10; i++) {
            channel.basicPublish("exchangeFanout","",null,("你好，MQ"+i).getBytes());
        }
        //8.关闭资源
        channel.close();
        connection.close();
    }
}

2、站内信消费者(其他同理)

package com.zj.mq.publish;
import com.rabbitmq.client.*;
import com.sun.deploy.ui.AboutDialog;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*站内信消费者*/
public class ConsumerStation {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory connectionaFactory = new ConnectionFactory();
        connectionaFactory.setHost("192.168.66.100");
        connectionaFactory.setPort(5672);
        connectionaFactory.setUsername("MQzhang");
        connectionaFactory.setPassword("MQzhang");
        connectionaFactory.setVirtualHost("/");
        //2.创建连接
        Connection connection = connectionaFactory.newConnection();
        //3.创建信道
        Channel channel = connection.createChannel();
        //4.监听队列（一直在连接不会关闭连接）
        /*
         * 参数一：监听的队列名
         * 参数二：是否自动签收（消费完消息后自动告诉MQ消息消费完了），如果设置为false需要手动确认消息，否则MQ会一直发送消息。
         * 参数三：Consumer的实现类，重写该类方法表示接受到消息后如何消费,body就是消息的字节数组。
         * */
        channel.basicConsume("stationQueue",true,new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("发送站内信:"+message);
            }
        });
    }
}

发送站内信:你好，MQ0
发送站内信:你好，MQ1
发送站内信:你好，MQ2
发送站内信:你好，MQ3
发送站内信:你好，MQ4
发送站内信:你好，MQ5
发送站内信:你好，MQ6
发送站内信:你好，MQ7
发送站内信:你好，MQ8
发送站内信:你好，MQ9

当然也能创建多个消费者来监听同一个队列来提高消费速度。

3.4 路由模式

使用发布订阅模式时，所有消息都会发送到绑定的队列中（发送到绑定到交换机上的每个队列，队列再发送给消费者），但很多时候，不是所有消息都无差别的发布到所有队列中。比如电商网站的促销活动，双十一大促可能会发布到所有队列；而一些小的促销活动为了节约成本，只发布到站内信队列。此时需要使用路由模式(Routing)完成这一需求。

特点

每个队列绑定路由关键字RoutingKey
生产者将带有RoutingKey的消息发送给交换机，交换机根据RoutingKey转发到指定队列。路由模式使用direct交换机。

编写生产者

package com.zj.mq.route;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*生产者*/
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("MQzhang");
        connectionFactory.setPassword("MQzhang");
        connectionFactory.setVirtualHost("/");
        //2.创建连接
        Connection connection = connectionFactory.newConnection();
        //3.建立信道
        Channel channel = connection.createChannel();
        //4.创建交换机fanout
        /*
        * 参数一：交换机名称
        * 参数二：交换机类型
        * 参数三： 交换机是否持久化（关闭控制台是否还存在）*/
        channel.exchangeDeclare("exchangeRoute", BuiltinExchangeType.DIRECT,false);
        //5.创建三个队列(分别模拟邮件发送、短信发送、站内信发送)
        /* 参数1：队列名
         * 参数2：是否持久化，true表示MQ重启后队列还在。
         * 参数3：是否私有化，false表示所有消费者都可以访问，true表示只有第一次拥有它的消费者才能访问
         * 参数4：是否自动删除，true表示不再使用队列时自动删除队列
         * 参数5：其他额外参数
         * */
        channel.queueDeclare("mailQueue", false,false,false,null);
        channel.queueDeclare("messageQueue", false,false,false,null);
        channel.queueDeclare("stationQueue", false,false,false,null);
        //6.交换机绑定队列
        /*
        * 参数一：队列名称
        * 参数二：交换机名称
        * 参数三：路由关键字,一个队列可以有多个路由关键字
        * */
        channel.queueBind("mailQueue","exchangeRoute","import");
        channel.queueBind("messageQueue","exchangeRoute","normal");
        channel.queueBind("stationQueue","exchangeRoute","import");
        //7.往交换机发送消息,路由关键字是import，表示交换机会将消息发送到带有import关键字的队列。
        channel.basicPublish("exchangeRoute","import",null,("你好，import MQ").getBytes());
        //路由关键字是normal表示交换机会将消息发送到带有normal关键字的队列
        channel.basicPublish("exchangeRoute","normal",null,("你好，normal MQ").getBytes());
        //8.关闭资源
        channel.close();
        connection.close();
    }
}

编写消费者

消费者还是和其他模式的消费者是一样的。这里以mailQuene举例子。

package com.zj.mq.route;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*消费者*/
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("MQzhang");
        connectionFactory.setPassword("MQzhang");
        connectionFactory.setVirtualHost("/");
        //2.创建连接
        Connection connection = connectionFactory.newConnection();
        //3.创建信道
        Channel channel = connection.createChannel();
        //4.监听队列（一直在连接不会关闭连接）
        /*
        * 参数一：监听的队列名
        * 参数二：是否自动签收（消费完消息后自动告诉MQ消息消费完了），如果设置为false需要手动确认消息，否则MQ会一直发送消息。
        * 参数三：Consumer的实现类，重写该类方法表示接受到消息后如何消费,body就是消息的字节数组。
        *
        */
        channel.basicConsume("mailQueue",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("接受消息为："+message);
            }
        });
    }
}

3.5 通配符模式

通配符模式(Topic)是在路由模式的基础上，给队列绑定带通配符的路由关键字，只要消息的RoutingKey能实现通配符匹配，就会将消息转发到该队列。通配符模式比路由模式更灵活，使用topic交换机。

通配符规则：

消息设置RoutingKey时，RoutingKey由多个单词构成，中间以.分割。
队列设置RoutingKey时，#可以匹配任意多个单词，*可以匹配任意一个单词。

编写生产者

package com.zj.mq.topic;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*生产者*/
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("MQzhang");
        connectionFactory.setPassword("MQzhang");
        connectionFactory.setVirtualHost("/");
        //2.创建连接
        Connection connection = connectionFactory.newConnection();
        //3.建立信道
        Channel channel = connection.createChannel();
        //4.创建交换机fanout
        /*
        * 参数一：交换机名称
        * 参数二：交换机类型
        * 参数三： 交换机是否持久化（关闭控制台是否还存在）*/
        channel.exchangeDeclare("exchangeTopic", BuiltinExchangeType.TOPIC,false);
        //5.创建三个队列(分别模拟邮件发送、短信发送、站内信发送)
        channel.queueDeclare("mailQueue", false,false,false,null);
        channel.queueDeclare("messageQueue", false,false,false,null);
        channel.queueDeclare("stationQueue", false,false,false,null);
        //6.交换机绑定队列
        /*
        * 参数一：队列名称
        * 参数二：交换机名称
        * 参数三：路由关键字，【#.mail.#】 表示：mail前后可以匹配多个单词*/
        channel.queueBind("mailQueue","exchangeTopic","#.mail.#");
        channel.queueBind("messageQueue","exchangeTopic","#.message.#");
        channel.queueBind("stationQueue","exchangeTopic","#.station.#");
        //7.往交换机发送消息到三个队列
        channel.basicPublish("exchangeTopic","mail.message.station",null,("你好，MQ").getBytes());
        //8.关闭资源
        channel.close();
        connection.close();
    }
}

编写消费者

也是和其他模式的消费者是一样的只需要监听消费者。

package com.zj.mq.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*消费者*/
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.66.100");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("MQzhang");
        connectionFactory.setPassword("MQzhang");
        connectionFactory.setVirtualHost("/");
        //2.创建连接
        Connection connection = connectionFactory.newConnection();
        //3.创建信道
        Channel channel = connection.createChannel();
        //4.监听队列（一直在连接不会关闭连接）
        /*
        * 参数一：监听的队列名
        * 参数二：是否自动签收（消费完消息后自动告诉MQ消息消费完了），如果设置为false需要手动确认消息，否则MQ会一直发送消息。
        * 参数三：Consumer的实现类，重写该类方法表示接受到消息后如何消费,body就是消息的字节数组。
        *
        */
        channel.basicConsume("mailQueue",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("接受消息为："+message);
            }
        });
    }
}

四、SpringBoot整合RabbitMQ

之前我们使用原生JAVA操作RabbitMQ较为繁琐，接下来我们使用SpringBoot整合RabbitMQ，简化代码编写。

1.创建SpringBoot项目，引入RabbitMQ起步依赖（springboot版本是2.7.0）

<!-- RabbitMQ起步依赖 -->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.编写配置文件

spring:
  rabbitmq:
    host: 192.168.66.100
    port: 5672
    username: MQzhang
    password: MQzhang
    virtual-host: /
#日志格式
logging:
  pattern:
    console: '%d{HH:mm:ss.SSS} %clr(%-5level) ---  [%-15thread] %cyan(%-50logger{50}):%msg%n'

3.创建队列和交换机

SpringBoot整合RabbitMQ时，需要在配置类创建队列和交换机，写法如下：

package com.zj.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
    private final String EXCHANGE_NAME = "boot_topic_exchange";
    private final String QUEUE_NAME = "boot_queue";
    // 创建交换机
    @Bean(EXCHANGE_NAME)
    public Exchange getExchange() {
        return ExchangeBuilder
                .topicExchange(EXCHANGE_NAME) // 交换机类型和名称
                .durable(true) // 是否持久化
                .build();
    }
    // 创建队列
    @Bean(QUEUE_NAME)
    public Queue getMessageQueue() {
        return new Queue(QUEUE_NAME); // 队列名
    }
    // 交换机绑定队列
    @Bean
    public Binding bindMessageQueue(@Qualifier(EXCHANGE_NAME) Exchange exchange,
                                    @Qualifier(QUEUE_NAME) Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("#.message.#")
                .noargs();
    }
}

高性能消息中间件 RabbitMQ（三）

3.3 发布订阅模式

特点

1、编写生产者

2、站内信消费者(其他同理)

3.4 路由模式

特点

编写生产者

编写消费者

3.5 通配符模式

编写生产者

编写消费者

四、SpringBoot整合RabbitMQ

1.创建SpringBoot项目，引入RabbitMQ起步依赖（springboot版本是2.7.0）

2.编写配置文件

3.创建队列和交换机

热门文章

最新文章

相关课程

相关电子书

相关实验场景