RabbitMQ（二）：Java-工作队列模式

痛定思痛。 2021-09-24 15:30 383阅读 0赞

RabbitMQ系列二

### 文章目录 ###

*   *  工作队列模式(work queue)
     *  引入pom
     *  生产者
     *  消费者1
     *  消费者2

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

## 工作队列模式(work queue) ##

> 一个消息轮流被接收

![工作队列模式][20200903092346943.png_pic_center]  
特点：

*  一个生产者发送消息
 *  多个消费者共同监听一个队列的消息
 *  消息不能被重复消费
 *  采用**轮询**的方式将消息平均发送给消费者(c1,c2,c1,c2…交替)

## 引入pom ##

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
    </dependency>

## 生产者 ##

public class Send { 
    	// 队列名
        private final static String QUEUE_NAME = "FIRST-MQ";
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 建立新的连接
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("106.**.**.82");
            factory.setPort(5672);
            // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = null;
            Channel channel = null;
            try{ 
                connection = factory.newConnection();
                // 创建内部会话通道，生产者和mq服务通信都在channel中完成
                channel = connection.createChannel();
                // 声明交换机，这里暂时没有,就会用默认的交换机
    
                // 声明创建队列，如果没有，则创建
                // 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
                /** * queue：队列名称 * durable：是否持久化，存到磁盘中 * exclusive:是否排他，是否独占连接，该队列只允许在这个connection中访问，如果关闭连接，则删除队列 * autoDelete:不再使用时自动删除。和exclusive 一同设为true 可以实现临时队列。 * args：扩展参数，比如存活时间 */
                channel.queueDeclare(QUEUE_NAME,false,false,false, null);
                String message = "work queue 模式，消息来了。";
                // 发送消息
                // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
                /** * exchange:交换机 如果不指定("")，就默认交换机 * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称 * props:额外属性 * body：消息内容 */
                channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
                System.out.println("send message:"+message);
            } catch (TimeoutException | IOException e) { 
                e.printStackTrace();
            } finally { 
                // 先关闭通道
                channel.close();
                // 再关闭连接
                connection.close();
            }
        }
    }

## 消费者1 ##

public class Recv { 
        private final static String QUEUE_NAME = "FIRST-MQ";
        public static void main(String[] args) { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("106.**.**.82");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            try{ 
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
                // 声明队列
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                System.out.println(" [*] Waiting for messages.");
    
                // 定义实现消费方法
                Consumer consumer = new DefaultConsumer(channel) { 
                    /** * 当接收到消息后，此方法将被调用 * @description: * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等 * @param properties:消息的属性，发送消息时 设置的额外属性 * @param body :消息内容 */
                    @SneakyThrows
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        String message = new String(body, StandardCharsets.UTF_8);
                        System.out.println("消费者1 receive:" + message);
                        TimeUnit.SECONDS.sleep(1);
                        // 手动回复ack
                        // channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                };
                // 接收消息 监听队列
                // 参数：String queue, boolean autoAck, Consumer callback
                /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */
                channel.basicConsume(QUEUE_NAME, true, consumer);
            } catch (IOException | TimeoutException | InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

## 消费者2 ##

public class Recv2 { 
        private final static String QUEUE_NAME = "FIRST-MQ";
        public static void main(String[] args) { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("106.**.**.82");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            try{ 
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
                // 声明队列
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                System.out.println(" [*] Waiting for messages.");
    
                // 定义实现消费方法
                Consumer consumer = new DefaultConsumer(channel) { 
                    /** * 当接收到消息后，此方法将被调用 * @description: * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等 * @param properties:消息的属性，发送消息时 设置的额外属性 * @param body :消息内容 */
                    @SneakyThrows
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        String message = new String(body, StandardCharsets.UTF_8);
                        System.out.println("消费者2 receive:" + message);
                        TimeUnit.SECONDS.sleep(1);
                        // 手动回复ack
                        // channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                };
                // 接收消息 监听队列
                // 参数：String queue, boolean autoAck, Consumer callback
                /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */
                channel.basicConsume(QUEUE_NAME, true, consumer);
            } catch (IOException | TimeoutException | InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

先启动两个消费者，然后再多次启动生产者，就能看到效果啦。

[20200903092346943.png_pic_center]: /images/20210923/162f88cf707e4e54bb67a6268c137cea.png