RabbitMQ —— 工作模式

r囧r小猫 2023-02-12 14:24 31阅读 0赞

# 前文 #

[消息中间件 —— 简介][Link 1]

[RabbitMQ —— 介绍][RabbitMQ _]

[RabbitMQ —— 下载、安装][RabbitMQ _ 1]

### 文章目录 ###

*  前文
 *  发布订阅
 *   *  消息生产者
     *  消息消费者
     *  测试
 *  Routing 模式
 *   *  Routing 模式和发布订阅模式的区别？
     *  消息生产者
     *  消息消费者
     *  测试
 *  Topic 模式
 *   *  Topic 与 Routing 的区别？
     *  消息生产者
     *  消息消费者
     *  测试

# 发布订阅 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70]  
1、一个生产者将消息发送给交换机  
2、与交换机绑定的有多个队列，每个消费者监听自己的队列  
3、生产者将消息发送给交换机，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接受到消息  
4、如果消息发给没有绑定队列的交换机上消息将丢失

## 消息生产者 ##

package com.java.rabbitmq.public_subject;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    
    /** * @author Woo_home * @create 2020/5/26 14:45 */
    
    public class Producer_Public { 
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] args)
        { 
            try
            { 
                // 获取连接
                Connection connection = ConnectionUtils.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
                // 声明交换机（分发:发布/订阅模式）
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
                // 发送消息
                for (int i = 0; i < 10; i++)
                { 
                    String message = "this is user message" + i;
                    System.out.println("[send]：" + message);
                    //发送消息
                    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8"));
                    Thread.sleep(5 * i);
                }
                channel.close();
                connection.close();
            }
            catch (Exception e)
            { 
                e.printStackTrace();
            }
        }
    }

## 消息消费者 ##

package com.java.rabbitmq.public_subject;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/26 15:12 */
    
    public class Consumer_subscribe_email { 
    
        //交换机名称
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        //队列名称
        private static final String QUEUE_NAME    = "test_queue_email";
    
        public static void main(String[] args) throws Exception { 
            try
            { 
    
                // 获取连接
                Connection connection = ConnectionUtils.getConnection();
                // 创建通道
                final Channel channel = connection.createChannel();
                // 声明交换机（分发:发布/订阅模式）
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
                // 声明队列
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                // 将队列绑定到交换机
                channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
                // 保证一次只分发一个
                int prefetchCount = 1;
                channel.basicQos(prefetchCount);
                
    			System.out.println("监听消息-----------------------------");
    
                // 定义消费者
                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, "utf-8");
                        System.out.println("[email] Receive message：" + message);
                        try { 
                            // 消费者休息2s处理业务
                            Thread.sleep(1000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            System.out.println("[1] done");
                            // 手动应答
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                // 设置手动应答
                boolean autoAck = false;
                // 监听队列
                channel.basicConsume(QUEUE_NAME, autoAck, consumer);
            }
            catch (IOException e)
            { 
                e.printStackTrace();
            }
        }
    }

package com.java.rabbitmq.public_subject;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/26 15:12 */
    
    public class Consumer_subscribe_sms { 
    
        //交换机名称
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        //队列名称
        private static final String QUEUE_NAME    = "test_queue_phone";
    
        public static void main(String[] args) throws Exception { 
            try
            { 
    
                // 获取连接
                Connection connection = ConnectionUtils.getConnection();
                // 创建通道
                final Channel channel = connection.createChannel();
                // 声明交换机（分发:发布/订阅模式）
                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
                // 声明队列
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                // 将队列绑定到交换机
                channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
                // 保证一次只分发一个
                int prefetchCount = 1;
                channel.basicQos(prefetchCount);
    		
    		    System.out.println("监听消息-----------------------------");
    
    
                // 定义消费者
                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, "utf-8");
                        System.out.println("[phone] Receive message：" + message);
                        try
                        { 
                            // 消费者休息2s处理业务
                            Thread.sleep(1000);
                        }
                        catch (InterruptedException e)
                        { 
                            e.printStackTrace();
                        }
                        finally
                        { 
                            System.out.println("[2] done");
                            // 手动应答
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                // 设置手动应答
                boolean autoAck = false;
                // 监听队列
                channel.basicConsume(QUEUE_NAME, autoAck, consumer);
            }
            catch (IOException e)
            { 
                e.printStackTrace();
            }
        }
    }

## 测试 ##

先启动两个消费端  
![在这里插入图片描述][20200526214640938.png]  
然后启动生产者端  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 1]  
这样两个消费端都可以接收到消息了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 3]

# Routing 模式 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 4]  
路由模式：

1、一个交换机绑定多个队列，每个队列设置 routingKey，并且一个队列可以设置多个 routingKey  
2、每个消费者监听自己的队列  
3、生产者将消息发给交换机，发送消息时需要指定 routingKey 的值，交换机来判断该 routingKey 的值和哪个队列的 routingKey 相等，如果相等则将消息转发给该队列

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

简单来说就是只发给指定的队列，其它的队列都收不到

## Routing 模式和发布订阅模式的区别？ ##

只不过是使用的交换机不同

*  **发布 / 订阅模式 Publish / Subscribe** 使用交换机类型为 **fanout**
 *  **路由模式 Routing** 使用交换机类型为 **direct**
 *  **通配符模式 Topics** 使用交换机类型为 **topic**

## 消息生产者 ##

package com.java.rabbitmq.direct;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * @author Woo_home * @create 2020/5/27 11:01 */
    
    public class TestProducer { 
    
        private static final String QUEUE_NAME = "direct_queue";
    
        public static void main(String[] args) throws Exception { 
    
            // 创建连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            for (int i = 0; i < 100; i++) { 
                String message = "direct 消息 " + i;
                // 发送消息到队列中
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes("utf-8"));
                System.out.println("发送消息：" + message);
            }
            // 关闭通道
            channel.close();
            connection.close();
        }
    }

## 消息消费者 ##

这里有两个消费者

package com.java.rabbitmq.direct;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/27 11:02 */
    
    public class TestConsumer1 { 
    
        private static final String QUEUE_NAME = "direct_queue";
    
        public static void main(String[] args) throws Exception { 
    
            // 创建连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            // 声明要关注的队列
            channel.queueDeclare(QUEUE_NAME, false, false, true, null);
    
            System.out.println("[1]队列正在监听消息：");
    
            Consumer consumer = 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("[1] 接收到消息：" + message);
                }
            };
    
            // 自动回复队列应答，RabbitMQ 中的消息确认机制
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

其实两个消费者的代码是基本一样的

package com.java.rabbitmq.direct;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/27 11:02 */
    
    public class TestConsumer2 { 
    
        private static final String QUEUE_NAME = "direct_queue";
    
        public static void main(String[] args) throws Exception { 
    
            // 创建连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            // 声明要关注的队列
            channel.queueDeclare(QUEUE_NAME, false, false, true, null);
    
            System.out.println("[2]队列正在监听消息：");
    
            Consumer consumer = 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("[2] 接收到消息：" + message);
                }
            };
    
            // 自动回复队列应答，RabbitMQ 中的消息确认机制
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

## 测试 ##

先启动两个消费者  
![在这里插入图片描述][20200527111729216.png]  
![在这里插入图片描述][20200527111741611.png]  
然后再启动生产者（图有点长，只截取一部分图）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 5]  
然后在观察两个消费者的输出  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 7]

# Topic 模式 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dvb19ob21l_size_16_color_FFFFFF_t_70 8]

## Topic 与 Routing 的区别？ ##

Topic 和 Routing 的基本原理相同，即：生产者将消息发送给交换机，交换机根据 routingKey 将消息转发给与 routingKey 匹配的队列

不同之处是：routingKey 的匹配方式，Routing 模式是相等匹配，而 Topic 模式是统配符匹配

符号#：匹配一个或者多个词，比如 inform.# 可以匹配 inform.sms、inform.email、inform.email.sms
    
    符号：只能匹配一个词，比如 inform.* 可以匹配 inform.sms、inform.email

## 消息生产者 ##

package com.java.rabbitmq.topic;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * @author Woo_home * @create 2020/5/27 13:07 */
    
    public class TopicProducer { 
    
        // 队列名称
        public static final String EXCHANGE_NAME = "topics_exchange";
    
        public static void main(String[] args) throws Exception { 
    
            // 创建连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            // 声明交换机以及类型
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    
            String[] routing_keys = new String[]{ "abc.news", "abc.weather", "efg.news", "efg.weather"};
            String[] messages = new String[]{ "abc新闻", "abc天气", "efg新闻", "efg天气"};
    
            for (int i = 0; i < routing_keys.length; i++) { 
                String routingKey = routing_keys[i];
                String message = messages[i];
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
                System.out.printf("发送到路由的消息是 ：%s，内容是：%s%n ", routingKey, message);
            }
    
            // 关闭通道
            channel.close();
            connection.close();
        }
    }

## 消息消费者 ##

模拟两个消费者，代码基本是一样的，只是接收的消息有所不同

package com.java.rabbitmq.topic;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/27 13:33 */
    
    public class TestConsumerABC { 
    
        // 队列名称
        public static final String EXCHANGE_NAME = "topics_exchange";
    
        public static void main(String[] args) throws Exception { 
    
            // 创建连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            // 声明交换机以及类型
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    
            // 获取一个临时队列
            String queueName = channel.queueDeclare().getQueue();
    
            // 接收 abc 消息
            channel.queueBind(queueName, EXCHANGE_NAME, "abc.*");
    
            System.out.println("[1]队列等待接收消息...");
    
            Consumer consumer = 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("[1]队列接收到消息 " + message);
                }
            };
            // 自动应答机制
            channel.basicConsume(queueName, true, consumer);
        }
    }

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

package com.java.rabbitmq.topic;
    
    import com.java.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * @author Woo_home * @create 2020/5/27 13:37 */
    
    public class TestConsumerEFG { 
    
        // 队列名称
        public static final String EXCHANGE_NAME = "topics_exchange";
    
        public static void main(String[] args) throws Exception { 
    
            // 创建连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            // 声明交换机以及类型
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    
            // 获取一个临时队列
            String queueName = channel.queueDeclare().getQueue();
    
            // 接收 efg 信息
            channel.queueBind(queueName, EXCHANGE_NAME, "*.news");
    
            System.out.println("[2]队列等待接收消息...");
    
            Consumer consumer = 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("[2]接收到消息 " + message);
                }
            };
    
            // 自动应答机制
            channel.basicConsume(queueName, true, consumer);
        }
    }

## 测试 ##

先启动两个消费者  
![在这里插入图片描述][2020052713410521.png]  
![在这里插入图片描述][20200527134117605.png]  
然后再启动消息生产者，可以发现，生产者已经将信息发送出去  
![在这里插入图片描述][20200527134141424.png]  
消息消费者 1 接受到的信息  
![在这里插入图片描述][2020052713421810.png]  
消息消费者 2 接受到的信息  
![在这里插入图片描述][20200527134150388.png]

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