RabbitMq（二）五种队列模式

向右看齐 2023-07-24 09:24 23阅读 0赞

## 一、simple简单队列 ##

![在这里插入图片描述][2020041022160013.png]

*  消息产生着将消息放入队列
 *  消息的消费者(consumer) 监听(while) 消息队列,如果队列中有消息,就消费掉,消息被拿走后,自动从队列中删除(隐患  
    消息可能没有被消费者正确处理,已经从队列中消失了,造成消息的丢失)应用场景:聊天(中间有一个过度的服务器;p端,c端)

#### 生产者 ####

//生产者推送消息到队列
    public class PubEmail {
        //将对象序列化为字节数组
        public static byte[] ser(Object obj) throws IOException {
            //接收被写入的字节数组
            ByteArrayOutputStream bos= new ByteArrayOutputStream();
            //把对象序列化成字节数组
            ObjectOutputStream oos= new ObjectOutputStream(bos);
            //写入
            oos.writeObject(obj);
            return bos.toByteArray();
        }
        //反序列化
        public static Object dser(byte[] src) throws Exception{
            //从字节数组读取数据
            ByteArrayInputStream bis = new ByteArrayInputStream(src);
            //把字节数组反序列化成对象
            ObjectInputStream ois= new ObjectInputStream(bis);
            return ois.readObject();
        }
    
        //任务被发送的队列名称
        static String QUEUE_NAME="MAIL";
        public static void main(String[] args) throws Exception {
            //模拟一个任务消息,以发邮件为例
            Map map = new HashMap();
            //接收邮件的邮箱
            map.put("sendto", "241243");
            //邮件标题
            map.put("subject", "测试邮件");
            //邮件内容
            map.put("content", "注册成功你的验证码是135469");
            //连接远程rabbit-server服务器
            ConnectionFactory factory = new ConnectionFactory();
            //本机LinuxIP
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            //获取连接
            Connection connection = factory.newConnection();
            //从连接中创建通道
            Channel channel = connection.createChannel();
            //定义创建一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //发送消息
            channel.basicPublish("", QUEUE_NAME, null, ser(map)); //注意发送和接受段相同字符集否则出现乱码
            channel.close();
            connection.close();
        }
    
    }

#### 消费者 ####

//消费者接收从队列数据
    public class Recv {
        //将对象序列化为字节数组
        public static byte[] ser(Object obj) throws IOException {
            //接收被写入的字节数组
            ByteArrayOutputStream bos= new ByteArrayOutputStream();
            //把对象序列化成字节数组
            ObjectOutputStream oos= new ObjectOutputStream(bos);
            //写入
            oos.writeObject(obj);
            return bos.toByteArray();
        }
        //反序列化
        public static Object dser(byte[] src) throws Exception{
            //从字节数组读取数据
            ByteArrayInputStream bis = new ByteArrayInputStream(src);
            //把字节数组反序列化成对象
            ObjectInputStream ois= new ObjectInputStream(bis);
            return ois.readObject();
        }
    
        //任务被发送的队列名称
        static String QUEUE_NAME="MAIL";
        public static void main(String[] args)  throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            //本机LinuxIP
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            //获取连接
            Connection connection = factory.newConnection();
            //从连接中创建通道
            Channel channel = connection.createChannel();
            //定义创建一个队列（保证有队列，如果确定一定存在，可以省略）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            //监听队列
            channel.basicConsume(QUEUE_NAME,true,consumer);
            //如果没有会阻塞等待，生产者一发送就会那到
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            Map map = (Map) dser(delivery.getBody());
            System.out.println(map.get("subject"));
        }
    }

## 二、work工作模式 ##

![在这里插入图片描述][20200410221931368.png]

*  消息产生者将消息放入队列消费者可以有多个,消费者1,消费者2,同时监听同一个队列,消息被消费?C1  
    C2共同争抢当前的消息队列内容,谁先拿到谁负责消费消息(隐患,高并发情况下,默认会产生某一个消息被多个消费者共同使用,可以设置一个开关(syncronize,与同步锁的性能不一样)  
    保证一条消息只能被一个消费者使用)
 *  应用场景:红包;大项目中的资源调度(任务分配系统不需知道哪一个任务执行系统在空闲,直接将任务扔到消息队列中,空闲的系统自动争抢)

public class ConnectionUtil {
    
        public static Connection getConnection() throws Exception {
            //定义连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //设置服务地址
            factory.setHost("localhost");
            //端口
            factory.setPort(5672);
            //设置账号信息，用户名、密码、vhost
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 通过工程获取连接
            Connection connection = factory.newConnection();
            return connection;
        }
    
    }

#### 生产者 ####

public class Send {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            for (int i = 0; i < 100; i++) {
                // 消息内容
                String message = "" + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
    
                Thread.sleep(i * 10);
            }
    
            channel.close();
            connection.close();
        }
    }

#### 消费者1 ####

public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 同一时刻服务器只会发一条消息给消费者(能者多劳)
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                //休眠
                Thread.sleep(10);
                // 返回确认状态
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

#### 消费者2 ####

public class Recv2 {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 同一时刻服务器只会发一条消息给消费者
            //channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成状态
            channel.basicConsume(QUEUE_NAME, false, consumer);
            // 监听队列，自动返回完成状态
            channel.basicConsume(QUEUE_NAME, true, consumer);
            //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);不要这个
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                // 休眠1秒
                Thread.sleep(1000);
                //反馈消息的消息状态
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

#### 消费确认模式 ####

自动模式

// 监听队列，自动返回完成状态
            channel.basicConsume(QUEUE_NAME, true, consumer);
            //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);不要这个

手动模式

// 监听队列，手动返回完成状态
            channel.basicConsume(QUEUE_NAME, false, consumer);

## 三、publish/subscribe订阅模式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lNYW5peQ_size_16_color_FFFFFF_t_70]  
1、1个生产者，多个消费者  
2、每一个消费者都有自己的一个队列  
3、生产者没有将消息直接发送到队列，而是发送了交换机  
4、每个队列都要绑定到交换机  
5、生产者发送的消费，经过交换机，到达队列，实现，一个消息被多个消费者获取的目的。  
注：如果交换器没有绑定队列，数据会丢失，交换机没有存储消息的能力，消息只能存在在队列中。  
相关场景:邮件群发,群聊天,广播(广告)

#### 生产者 ####

public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明交换机exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            // 消息内容
            String message = "Hello World!商品已经更新";
            channel.basicPublish(EXCHANGE_NAME, "key", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

#### 消费者1 ####

public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

#### 消费者2 ####

public class Recv2 {
    
        private final static String QUEUE_NAME = "test_queue_work2";
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

## 四、routing路由模式 ##

![在这里插入图片描述][20200411215316949.png]  
如三、把不同消费者key2改动一下

// 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2");

生产者指定 key2

channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes());

生产者就只给key2的消费者发消息。

*  消息生产者将消息发送给交换机按照路由判断,路由是字符串(info)  
    当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息;
 *  根据业务功能定义路由字符串
 *  从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中业务场景:error  
    通知;EXCEPTION;错误通知的功能;传统意义的错误通知;客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误;

## 五、topic 通配符模式（路由模式的一种)） ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lNYW5peQ_size_16_color_FFFFFF_t_70 1]  
如四，key中可以加通配符\*\#如：key\*

1.  星号井号代表通配符
2.  星号代表多个单词,井号代表一个单词
3.  路由功能添加模糊匹配
4.  消息产生者产生消息,把消息交给交换机
5.  交换机根据key的规则模糊匹配到对应的队列,由队列的监听消费者接收消息消费

[2020041022160013.png]: /images/20230528/87e762c6ceb54db59698f24785348630.png
[20200410221931368.png]: /images/20230528/f0b2aad301a948a3a97e90b93b1b944e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lNYW5peQ_size_16_color_FFFFFF_t_70]: /images/20230528/4e86ab0d3b5347d187c302946f94a806.png
[20200411215316949.png]: /images/20230528/a6876a493bab414c86b7e96353ab4a60.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lNYW5peQ_size_16_color_FFFFFF_t_70 1]: /images/20230528/c262ae5b7e0a48749cc24bbe55a687d9.png