RabbitMQ学习笔记（三）工作模式及相应代码

ゞ浴缸里的玫瑰 2023-01-14 06:56 177阅读 0赞

## **一、简单模式（simple）** ##

（1）模型：  
![在这里插入图片描述][20210420112129656.png]

*  P：生产者，也就是要发送消息的程序
 *  C：消费者：消息的接受者，会一直等待消息到来，这里可以设置成手动的ack,但如果设置成手动ack，处理完后要及时发送ack消息给队列，否则会造成内存溢出)。
 *  queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息

**生产者代码：**

//创建连接工厂
            //创建连接mq的连接工厂对象
            ConnectionFactory  connectionFactory = new ConnectionFactory();
     
            //设置连接rabbitmq主机
            connectionFactory.setHost("10.30.126.252");
            //设置端口号
            connectionFactory.setPort(5672);
            //设置访问虚拟主机的用户名和密码
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            //设置连接那个虚拟主机
            connectionFactory.setVirtualHost("/");
            //获取连接对象
            Connection connection = connectionFactory.newConnection();
            //获取连接中通道
            Channel channel = connection.createChannel();
            //通道绑定对应消息队列
            //参数1:  队列名称 如果队列不存在自动创建
            //参数2:  用来定义队列特性是否要持久化 true 持久化队列   false 不持久化
            //参数3:  exclusive 是否独占队列  true 独占队列   false  不独占
            //参数4:  autoDelete: 是否在消费完成后自动删除队列  true 自动删除  false 不自动删除
            //参数5:  额外附加参数
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            String msg = "My first message";
            //发布消息
            //参数1: 交换机名称(简单模式下为空) 参数2:队列名称  参数3:传递息额外设置  参数4:消息的具体内容
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            channel.close();
            connection.close();

消费者代码：

public class Consumer {
        
        
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("10.30.126.252");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("admin");
            factory.setPassword("admin");
            Connection connection = factory.newConnection();
     
            //创建通道
            Channel channel = connection.createChannel();
     
            //通道绑定队列：与生产端一致
            channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
     
            //获取消息
            //参数1: 消费那个队列的消息 队列名称
            //参数2: 开始消息的自动确认机制[只要消费就从队列删除消息]
            //参数3: 消费时的回调接口
            channel.basicConsume(Producer.QUEUE_NAME, true, new DefaultConsumer(channel) {
        
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                    System.out.println("取出消息:===>" + new String(body));
                }
            });
        }
    }

## 二、工作模式（work） ##

（1）概念：Work queues，也被称为（Task queues），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。  
（2）模型：![在这里插入图片描述][20210420112818724.png]  
P：生产者：任务的发布者  
C1：消费者-1，领取任务并且完成任务，假设完成速度较慢  
C2：消费者-2：领取任务并完成任务，假设完成速度快

**生产者：**

public class Producer {
        
        
        public static final String QUEUE_NAME = "my_second_queue";
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            //创建连接mq的连接工厂对象
            Connection connection = RabbitMQUtils.getConnection();
            //获取连接中通道
            Channel channel = connection.createChannel();
            //通道绑定对应消息队列
            //参数1:  队列名称 如果队列不存在自动创建
            //参数2:  用来定义队列特性是否要持久化 true 持久化队列   false 不持久化
            //参数3:  exclusive 是否独占队列  true 独占队列   false  不独占
            //参数4:  autoDelete: 是否在消费完成后自动删除队列  true 自动删除  false 不自动删除
            //参数5:  额外附加参数
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            String msg = "My second message";
            //发布消息
            //参数1: 交换机名称(简单模式下为空) 参数2:队列名称  参数3:传递息额外设置  参数4:消息的具体内容
            for (int i = 0; i < 20; i++) {
        
                channel.basicPublish("", QUEUE_NAME, null, (i+1+"--"+msg).getBytes());
            }
            channel.close();
            connection.close();
        }
    }

**消费者：**

public class Consumer1 {
        
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
     
            //创建通道
            Channel channel = connection.createChannel();
     
            //通道绑定队列：与生产端一致
            channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
     
            //获取消息
            //参数1: 消费那个队列的消息 队列名称
            //参数2: 开始消息的自动确认机制[只要消费就从队列删除消息]
            //参数3: 消费时的回调接口
            channel.basicConsume(Producer.QUEUE_NAME, 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));
                }
            });
        }
    }、

（3）当有多个消费者时，我们的消息会被哪个消费者消费呢，我们又该如何均衡消费者消费信息的多少呢？

> 主要有两种模式：  
> 1、轮询模式的分发：一个消费者一条，按均分配；  
> 2、公平分发：根据消费者的消费能力进行公平分发，处理快的处理的多，处理慢的处理的少；按劳分配；  
> **注意点：**  
> （1）消费者一次接收一条消息，代码channel.BasicQos(0, 1, false);  
> （2） 公平分发需要消费者开启手动应答，关闭自动应答  
> 关闭自动应答代码channel.BasicConsume(“queue\_test”, false, consumer);  
> 消费者开启手动应答代码：channel.BasicAck(ea.DeliveryTag, false)

## 三、发布/订阅模式（publish/subscribe） ##

![在这里插入图片描述][20210420121956922.png]  
在广播模式下，消息发送流程是这样的：

*  可以有多个消费者
 *  每个消费者有自己的queue（队列）
 *  每个队列都要绑定到Exchange（交换机）
 *  生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
 *  交换机把消息发送给绑定过的所有队列
 *  队列的消费者都能拿到消息。实现一条消息被多个消费者消费

生产者：

public class Producer {
        
     
        public static final String EXCHANGE_NAME = "fanoutExchange";
        public static final String EXCHANGE_TYPE = "fanout";
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
            //获取连接中通道
            Channel channel = connection.createChannel();
            //声明交换机
            //参数1: 交换机名称
            //参数2: 交换机类型(fanout广播模式)
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
     
            String msg = "My third message----fanout";
            //发布消息
            //参数1: 交换机名称 参数2:队列名称  参数3:传递息额外设置  参数4:消息的具体内容
            channel.basicPublish(EXCHANGE_NAME, "", null, (msg).getBytes());
            channel.close();
            connection.close();
        }
    }

消费者：

public class Consumer1 {
        
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
     
            //创建通道
            Channel channel = connection.createChannel();
     
            //声明交换机
            //参数1: 交换机名称
            //参数2: 交换机类型(fanout广播模式)
            channel.exchangeDeclare(Producer.EXCHANGE_NAME,Producer.EXCHANGE_TYPE);
     
            //创建临时队列
            String queue = channel.queueDeclare().getQueue();
     
            //临时队列和交换机 [参数1:临时队列,参数2:交换机,参数3:路由]
            channel.queueBind(queue, Producer.EXCHANGE_NAME, "");
     
            //获取消息
            //参数1: 消费那个队列的消息 队列名称
            //参数2: 开始消息的自动确认机制[只要消费就从队列删除消息]
            //参数3: 消费时的回调接口
            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));
                }
            });
        }
    }

## 四、路由模式 ##

（1）模型：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk4MTMxNQ_size_16_color_FFFFFF_t_70]

*  P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
 *  X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
 *  C1：消费者，其所在队列指定了需要routing key 为 error 的消息
 *  C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

**生产者：**

public class Producer {
        
     
        public static final String EXCHANGE_NAME = "routeExchange";
        public static final String EXCHANGE_TYPE = "direct";
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
            //获取连接中通道
            Channel channel = connection.createChannel();
            //声明交换机
            //参数1: 交换机名称
            //参数2: 交换机类型(路由模式)
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
     
            String msg = "My fourth message----route";
     
            //发布消息[参数1:交换机名字,参数2:路由名字,参数3:消息内容]
            String key = "error";
            channel.basicPublish(EXCHANGE_NAME, key, null, ("发送给指定路由" + key + "的消息").getBytes());
            channel.close();
            connection.close();
        }
    }

消费者：

public class Consumer1 {
        
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
     
            //创建通道
            Channel channel = connection.createChannel();
     
            //声明交换机
            //参数1: 交换机名称
            //参数2: 交换机类型(fanout广播模式)
            channel.exchangeDeclare(Producer.EXCHANGE_NAME,Producer.EXCHANGE_TYPE);
     
            //创建临时队列
            String queue = channel.queueDeclare().getQueue();
     
            //临时队列和交换机并设置路由 [参数1:临时队列,参数2:交换机,参数3:路由]
            channel.queueBind(queue, Producer.EXCHANGE_NAME, "info");
            channel.queueBind(queue, Producer.EXCHANGE_NAME, "error");
            channel.queueBind(queue, Producer.EXCHANGE_NAME, "warn");
     
            //获取消息
            //参数1: 消费那个队列的消息 队列名称
            //参数2: 开始消息的自动确认机制[只要消费就从队列删除消息]
            //参数3: 消费时的回调接口
            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));
                }
            });
        }
    }

主题模式：  
![在这里插入图片描述][2021042012243531.png]

# 统配符
            * (star) can substitute for exactly one word.    匹配不多不少恰好1个词
            # (hash) can substitute for zero or more words.  匹配一个或多个词
    # 如:
            audit.#    匹配audit.irs.corporate或者 audit.irs 等
        	audit.*   只能匹配 audit.irs

生产者：

public class Producer {
        
     
        public static final String EXCHANGE_NAME = "topicExchange";
        public static final String EXCHANGE_TYPE = "topic";
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
            //获取连接中通道
            Channel channel = connection.createChannel();
            //声明交换机
            //参数1: 交换机名称
            //参数2: 交换机类型(路由模式)
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
     
            String msg = "My fifth message----topic";
            //发布消息[参数1:交换机名字,参数2:路由名字,参数3:消息内容]
            //使用动态路由
            String key = "user.save";
            channel.basicPublish(EXCHANGE_NAME, key, null, ("发送给指定路由" + key + "的消息:" + msg).getBytes());
            channel.close();
            connection.close();
        }
    }

消费者：

public class Consumer1 {
        
     
        public static void main(String[] args) throws IOException, TimeoutException {
        
     
            //创建连接工厂
            Connection connection = RabbitMQUtils.getConnection();
     
            //创建通道
            Channel channel = connection.createChannel();
     
            //声明交换机
            //参数1: 交换机名称
            //参数2: 交换机类型(fanout广播模式)
            channel.exchangeDeclare(Producer.EXCHANGE_NAME,Producer.EXCHANGE_TYPE);
     
            //创建临时队列
            String queue = channel.queueDeclare().getQueue();
     
            //临时队列和交换机并设置路由 [参数1:临时队列,参数2:交换机,参数3:路由]
            channel.queueBind(queue, Producer.EXCHANGE_NAME, "user.*");
     
            //获取消息
            //参数1: 消费那个队列的消息 队列名称
            //参数2: 开始消息的自动确认机制[只要消费就从队列删除消息]
            //参数3: 消费时的回调接口
            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));
                }
            });
        }
    }

[20210420112129656.png]: /images/20221022/31abf0286a824911b6cf17f39eb008a5.png
[20210420112818724.png]: /images/20221022/dfb1624b73534447939e9db3077e112f.png
[20210420121956922.png]: /images/20221022/0c76e5e91e6a45a681f9a476466f61ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mzk4MTMxNQ_size_16_color_FFFFFF_t_70]: /images/20221022/2a68ea46232045a49ff6f522464217fe.png
[2021042012243531.png]: /images/20221022/07ad2fe172ca4fe5a55572fb0affe197.png