RabbitMQ 工作队列模型

忘是亡心i 2023-01-15 16:18 147阅读 0赞

# 1 JAVA工程 #

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

**pom ：**

<dependencies>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.10.0</version>
            </dependency>
        </dependencies>
    
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.8.1</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>

# 2 RabbitMQ简单队列 #

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

## 2.1 Send ##

package com.rabbitmq.simple;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.nio.charset.StandardCharsets;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/23 09:45 */
    public class Send { 
    
    
        private static final String QUEUE_NAME = "hello";
    
    
        public static void main(String[] args) throws Exception { 
    
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.38.80");
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setPort(5672);
    
            //创建连接、创建信道
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
    
    
                /* * 队列名称 * 持久化配置，MQ重启该队列是否还在 * 独占，只能有一个消费者监听队列 * 自动删除：当没有消费者的时候，自动删除掉，一般是false * 其它参数 * * 注意：队列不存在则会自动创建，如果存在则不会被覆盖。 */
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
                //发送消息
                String msg = "Hello World";
                channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
                System.out.println("发送消息成功");
            }
    
    
        }
    
    
    }

## 2.2 Receive ##

package com.rabbitmq.simple;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.nio.charset.StandardCharsets;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/23 09:58 */
    public class Receive { 
    
    
        private static final String QUEUE_NAME = "hello";
    
    
        public static void main(String[] args) throws Exception { 
    
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.38.80");
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setPort(5672);
    
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //创建消费者
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    //会话名称
                    System.out.println("consumerTag = " + consumerTag);
                    //获取交换机、路由等信息
                    System.out.println("envelope = " + envelope);
                    System.out.println("properties = " + properties);
                    System.out.println("body = " + new String(body, StandardCharsets.UTF_8));
                }
    
    
            };
            //绑定消费者：队列、自动确认、消费者
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
        }
    
    
    }

## 2.3 启动消费者、生产者 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]

# 3 RabbitMQ 工作队列轮训策略 #

## 3.1 描述 ##

消息⽣产能⼒⼤于消费能⼒，增加多⼏个消费节点，和简单队列类似，增加多个⼏个消费节点，处于竞争关系。  
![在这里插入图片描述][2021042311080918.png]

## 3.2 Send ##

package com.rabbitmq.work;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.nio.charset.StandardCharsets;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/23 09:45 */
    public class Send { 
    
    
        private static final String QUEUE_NAME = "WORK";
    
    
        public static void main(String[] args) throws Exception { 
    
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.38.80");
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setPort(5672);
    
            //创建连接、创建信道
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
                //发送消息
                for (int i = 1; i <= 10; i++) { 
                    String msg = "Hello World" + i;
                    channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
                    System.out.println("发送消息" + i + "成功");
                }
            }
    
    
        }
    
    
    }

## 3.3 Receive ##

package com.rabbitmq.work;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.nio.charset.StandardCharsets;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/23 09:58 */
    public class Receive { 
    
    
        private static final String QUEUE_NAME = "WORK";
    
    
        public static void main(String[] args) throws Exception { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.38.80");
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setPort(5672);
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //创建消费者
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("receive = " + new String(body, StandardCharsets.UTF_8));
                    //手工确认
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
    
            //绑定消费者：队列、自动确认、消费者
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
        }
    
    
    }

## 3.4 启动2个消费者 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]

## 3.5 启动生产者 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]  
**描述：** 消费者1  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 6]  
**描述：** 消费者2  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 7]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]: /images/20221022/1fc475ee23394c20b038aa95d475112f.png
[20210423110542931.png]: /images/20221022/a0324f2374b94b4a8e157f0786d0fd59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]: /images/20221022/9969a969f2d44c5aad086651580403ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]: /images/20221022/fed2baffd5234fb59392ef893b780fe4.png
[2021042311080918.png]: /images/20221022/bd78a57df39e434eb21a5af2ddd1c03d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]: /images/20221022/6ad5a538caab490eb94ec6cc1683a6e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]: /images/20221022/5a7be537a3824190bb2822665ec69492.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]: /images/20221022/2d39974ef0b143c39739eea7f6a890fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 6]: /images/20221022/a416a7c4d4ab4c1f98d3d95c66033b8e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 7]: /images/20221022/0f5aaa9a48dd49b3b552ba08c156456e.png