rabbitmq系列（二）五种常见模式

朴灿烈づ我的快乐病毒、 2021-09-18 05:08 309阅读 0赞

## 一、简单模式 ##

**原理**：生产者将消息交给默认的交换机，交换机获取消息后交给绑定这个生产者的队列（投递规则为队列名称和routing key 相同的队列），监听当前队列的消费者获取信息并执行消费逻辑。

![简单模式.png][.png]

**场景**：有一个oa系统，用户通过接收手机验证码进行注册，页面上点击获取验证码后，将验证码放到消息队列，然后短信服务从队列中获取到验证码，并发送给用户。

**实现**：

生产者：

public class Producter { 
    
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
            // 4. 通过channel发布消息
            /** * 四个参数： * 第一个参数是交换机的名称 * 第二个参数是路由键 * 第三个参数标识消息的一些额外的属性 * 第四个是消息的具体的内容 */
            String message = "字节";
            for(int i = 0;i < 5;i   ){ 
    
                channel.basicPublish("","byte001",null,message.getBytes());
            }
            // 5. 释放资源，释放channel 和 链接对象
    
            channel.close();
            connection.close();
        }
    }

消费者：

public class Consumer { 
    
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            // 4. 创建出消息队列
            /** * 第一个参数是消息队列的名称 * 第二个参数表示消息是否持久化 * 第三个参数标识消息队列是否被channel独占 * 第四个参数标识是否自动删除消息队列，当消息队列没有绑定交换机后是否自动删除 * 第五个是消息队列扩展参数 */
            String queueName = "byte001";
            channel.queueDeclare(queueName, true, false, false, null);
    
            // 5. 创建消费者,对消息进行处理
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("获取到的消息：" s);
                }
            };
            // 6. 通过channel消费者和消息队列关联
            /** * 第一个参数是消息队列的名字 * 第二个参数是否自动签收（即消费消息后告知服务器已被消费） * 第三个参数是消费者 */
            channel.basicConsume(queueName, true, consumer);
        }
    }

## 二、工作模式 ##

**原理**：生产者将消息交给交换机，交换机交给绑定的队列，队列有多个消费者监听，一条消息只能由一个消费者消费，这样就形成了资源竞争，谁的资源空闲大，争抢到的可能性就大。

![工作模式.png][.png 1]

**场景**：有一个电商平台，有两个订单服务，用户下单的时候，任意一个订单服务消费用户的下单请求生成订单即可。不用两个订单服务同时消费用户的下单请求。

**实现**：

生产者：

public class Producter { 
    
        public static final String QUEUE_NAME = "byte002";
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            // 申明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            // 4. 通过channel发布消息
            /** * 四个参数： * 第一个参数是交换机的名称 * 第二个参数是路由键 * 第三个参数标识消息的一些额外的属性 * 第四个是消息的具体的内容 */
            String message = "字节";
            for(int i = 0;i < 100;i   ){ 
    
                channel.basicPublish("",QUEUE_NAME,null,(message i).getBytes());
            }
            // 5. 释放资源，释放channel 和 链接对象
    
            channel.close();
            connection.close();
        }
    }

消费者1：

public class Consumer { 
    
        public static final String QUEUE_NAME = "byte002";
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            // 4. 创建出消息队列
            /** * 第一个参数是消息队列的名称 * 第二个参数表示消息是否持久化 * 第三个参数标识消息队列是否被channel独占 * 第四个参数标识是否自动删除消息队列，当消息队列没有绑定交换机后是否自动删除 * 第五个是消息队列扩展参数 */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.basicQos(1); // 告诉服务器，在我们没有确认当前消息时不要给我们发送新的消息
            // 5. 创建消费者,对消息进行处理
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者1收到的内容：" s);
                    try { 
                        Thread.sleep(10); // 模拟消费耗时
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(),false); // 参数2false为确认收到消息，true为拒绝收到消息
                }
            };
            // 6. 通过channel消费者和消息队列关联
            /** * 第一个参数是消息队列的名字 * 第二个参数是否自动签收（即消费消息后告知服务器已被消费） * 第三个参数是消费者 */
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }

消费者2：

public class Consumer2 { 
    
        public static final String QUEUE_NAME = "byte002";
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            // 4. 创建出消息队列
            /** * 第一个参数是消息队列的名称 * 第二个参数表示消息是否持久化 * 第三个参数标识消息队列是否被channel独占 * 第四个参数标识是否自动删除消息队列，当消息队列没有绑定交换机后是否自动删除 * 第五个是消息队列扩展参数 */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.basicQos(1); // 告诉服务器，在我们没有确认当前消息时不要给我们发送新的消息
            // 5. 创建消费者,对消息进行处理
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者2收到的内容：" s);
                    try { 
                        Thread.sleep(500); // 模拟消费耗时
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(),false); // 参数2false为确认收到消息，true为拒绝收到消息
                }
            };
            // 6. 通过channel消费者和消息队列关联
            /** * 第一个参数是消息队列的名字 * 第二个参数是否自动签收（即消费消息后告知服务器已被消费） * 第三个参数是消费者 */
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }

保证资源竞争的代码就是这一行channel.basicQos(1);如果不加这一行，我们会发现两个消费者是轮询消费消息的。

## 三、发布订阅模式 ##

**原理**：生产者将消息扔给交换机，交换机类型是fanout，不同的队列注册到交换机上，不同的消费注册在不同的队列上。所有消费者都会收到消息。

![发布订阅模式.png][.png 2]

**场景**：有一个商城，我们新添加一个商品后，可能同时需要去更新缓存和数据库。

实现：

生产者：

public class Producter { 
    
        // 定义交换机的名字
        public static final String EXCHANGE_NAME="byte003";
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
            // 定义一个交换机，类型是fanout
            channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
    
            // 因为消息先发到交换机，交换机没有保存功能，所以如果没有消费者，消息会丢失
            channel.basicPublish(EXCHANGE_NAME,"",null,"发布订阅模式的消息".getBytes());
    
            channel.close();
            connection.close();
        }
    }

消费者1：

public class Consumer1 { 
        public static final String EXCHANGE_NAME="byte003";
    
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            String queueName = "queue003";
            channel.queueDeclare(queueName,false,false,false,null);
    
            // 绑定队列到交换机
            channel.queueBind(queueName,EXCHANGE_NAME,"");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者1：" s);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(queueName,false,consumer);
        }
    }

消费者2：

public class Consumer2 { 
        public static final String EXCHANGE_NAME="byte003";
    
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            String queueName = "queue004";
            channel.queueDeclare(queueName,false,false,false,null);
    
            // 绑定队列到交换机
            channel.queueBind(queueName,EXCHANGE_NAME,"");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者2：" s);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(queueName,false,consumer);
        }
    }

需要注意的一点就是交换机没有保存功能，如果没有消费者，则消息会丢失。

## 四、路由模式 ##

**原理**：生产者将消息发送给交换机，消息携带具体的routingkey。交换机类型是direct，接收到消息中的routingkey，比对与之绑定的队列的routingkey，分发到不同的队列上。

![路由模式.png][.png 3]

**场景**：还是一样，有一个商城，新添加了一个商品，实时性不是很高，只需要添加到数据库即可，不用刷新缓存。

**实现**：

生产者：

public class Producter { 
    
        // 定义交换机的名字
        public static final String EXCHANGE_NAME="byte004";
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
            // 定义一个交换机，类型是direct
            channel.exchangeDeclare(EXCHANGE_NAME,"direct");
    
            // 因为消息先发到交换机，交换机没有保存功能，所以如果没有消费者，消息会丢失
            channel.basicPublish(EXCHANGE_NAME,"key1",null,"发布路由模式的消息".getBytes());
    
            channel.close();
            connection.close();
        }
    }

消费者1：

public class Consumer1 { 
        public static final String EXCHANGE_NAME="byte004";
    
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            String queueName = "queue005";
            channel.queueDeclare(queueName,false,false,false,null);
    
            // 绑定队列到交换机
            /** * 参数3是routingkey，只有和它一样的routingkey的消息才会被当前消费者收到 */
            channel.queueBind(queueName,EXCHANGE_NAME,"key1");
            // 如果要接收多个routingkey的消息，在执行一次上面的代码即可，如下
            channel.queueBind(queueName,EXCHANGE_NAME,"key2");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者1：" s);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(queueName,false,consumer);
        }
    }

消费者2：

public class Consumer2 { 
        public static final String EXCHANGE_NAME="byte004";
    
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            String queueName = "queue006";
            channel.queueDeclare(queueName,false,false,false,null);
    
            // 绑定队列到交换机
            channel.queueBind(queueName,EXCHANGE_NAME,"key2");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者2：" s);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(queueName,false,consumer);
        }
    }

## 五、主题模式 ##

**原理**：路由模式的一种，路由功能添加了模糊匹配。星号（\*）代表1个单词，\#号（\#）代表一个或多个单词。具体可参考路由模式。

![主题模式.png][.png 4]

**场景**：还是一样，有一个商城，新添加了一个商品，实时性不是很高，只需要添加到数据库即可，数据库包含了主数据库mysql1和从数据库mysql2的内容，不用刷新缓存。

**实现**：

生产者：

public class Producter { 
    
        // 定义交换机的名字
        public static final String EXCHANGE_NAME="byte004";
        public static void main(String[] args) throws Exception { 
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
            // 定义一个交换机，类型是topic
            channel.exchangeDeclare(EXCHANGE_NAME,"topic");
    
            // 因为消息先发到交换机，交换机没有保存功能，所以如果没有消费者，消息会丢失
            channel.basicPublish(EXCHANGE_NAME,"key.1.2",null,"发布路由模式的消息".getBytes());
    
            channel.close();
            connection.close();
        }
    }

消费者1：

public class Consumer1 { 
        public static final String EXCHANGE_NAME="byte004";
    
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            String queueName = "queue005";
            channel.queueDeclare(queueName,false,false,false,null);
    
            // 绑定队列到交换机
            /** * 参数3是routingkey，只有和它一样的routingkey的消息才会被当前消费者收到 */
            channel.queueBind(queueName,EXCHANGE_NAME,"key.*");
            // 如果要接收多个routingkey的消息，在执行一次上面的代码即可，如下
            channel.queueBind(queueName,EXCHANGE_NAME,"abc.#");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者1：" s);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(queueName,false,consumer);
        }
    }

消费者2：

public class Consumer2 { 
        public static final String EXCHANGE_NAME="byte004";
    
        public static void main(String[] args) throws Exception { 
    
    
            // 1. 创建出链接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/");
    
            // 2. 通过链接工厂创建链接对象
            Connection connection = factory.newConnection();
    
            // 3. 通过链接对象创建出channel
            Channel channel = connection.createChannel();
    
            String queueName = "queue006";
            channel.queueDeclare(queueName,false,false,false,null);
    
            // 绑定队列到交换机
            channel.queueBind(queueName,EXCHANGE_NAME,"key.#");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                /** * consumerTag 用来标识.可以再监听队列时候设置 * envelope 信封,通过envelope可以通过这个获取到很多东西 * properties 额外的消息属性 * body:消息体 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    String s = new String(body, "UTF-8");
                    System.out.println("消费者2：" s);
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(queueName,false,consumer);
        }
    }

代码已上传：

github地址： [https://github.com/binzh303/zhixie-code-example][https_github.com_binzh303_zhixie-code-example]

gitee地址： [https://gitee.com/javaXiaoCaiJi/zhixie-code-example][https_gitee.com_javaXiaoCaiJi_zhixie-code-example]

> 如果文章对您有帮助，请记得点赞关注哟~ > 欢迎大家关注我的公众号：字节传说，每日推送技术文章供大家学习参考。

[.png]: /images/20210918/d8f7511a399c4fafafb0ee4cddbadc83.png
[.png 1]: /images/20210918/28b687f22a744405b06b93dcbd07a302.png
[.png 2]: /images/20210918/f5b187d53a5b4eb084655b0926d02643.png
[.png 3]: /images/20210918/aefddfe9d2e240d2a0105352d441707e.png
[.png 4]: /images/20210918/291f439b5404471a9a66bcce1102ad7e.png
[https_github.com_binzh303_zhixie-code-example]: https://github.com/binzh303/zhixie-code-example
[https_gitee.com_javaXiaoCaiJi_zhixie-code-example]: https://gitee.com/javaXiaoCaiJi/zhixie-code-example