1、Work queues 工作队列模式概念:
Work Queues 与入门程序的 简单模式 相比,多了一个或一些消费端,多个消费端共同消费同一个队列中的消息。
工作队列模式:在同一个队列中可以有多个消费者,消费者之间对于消息的接收是竞争关系。
Work Queues 与入门程序的 简单模式 的代码是几乎一样的;可以完全复制,并复制多一个消费者进行多个消费者同时消费消息的测试。
2、Work queues 工作队列模式 代码步骤:
1)生产者:发送30个消息
package com.study.rabbitmq.work;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
//work模式 发送消息
public class Producer {
static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
// 1. 创建连接工厂(设置RabbitMQ的连接参数);
ConnectionFactory connectionFactory= new ConnectionFactory();
//主机;默认localhost
connectionFactory.setHost("localhost");
//连接端口;默认5672
connectionFactory.setPort(5672);
//虚拟主机;默认/
connectionFactory.setVirtualHost("/");
//用户名;默认guest
connectionFactory.setUsername("guest");
//密码;默认guest
connectionFactory.setPassword("guest");
// 2. 创建连接
Connection connection = connectionFactory.newConnection();
// 3. 创建频道;
Channel channel = connection.createChannel();
// 4. 声明队列;
/**
* 参数1:队列名称
* 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
* 参数3:是否独占本连接
* 参数4:是否在不使用的时候队列自动删除
* 参数5:其它参数
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
for(int i=1;i<=30;i++){
//5. 发送消息(改动);
String message = "你好!work 模式---"+i;
/**
* 参数1:交换机名称;如果没有则指定空字符串(表示使用默认的交换机)
* 参数2:路由key,简单模式中可以使用队列名称
* 参数3:消息其它属性
* 参数4:消息内容
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("已发送消息:" + message);
}
// 6. 关闭资源
channel.close();
connection.close();
}
}
2)消费者:创建两个消费者监听同一个队列,查看两个消费者的接收消息是否存在重复。
package com.study.rabbitmq.work;
import com.rabbitmq.client.*;
import com.study.rabbitmq.util.ConnectionUtil;
import java.io.IOException;
/**
* work模式;消费者接收消息
*/
public class Consumer1 {
public static void main(String[] args) throws Exception {
// 1. 创建连接工厂;
// 2. 创建连接;(抽取一个获取连接的工具类)
Connection connection = ConnectionUtil.getConnection();
// 3. 创建频道;
final Channel channel = connection.createChannel();
// 4. 声明队列;
/**
* 参数1:队列名称
* 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
* 参数3:是否独占本连接
* 参数4:是否在不使用的时候队列自动删除
* 参数5:其它参数
*/
channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
//每次可以预取多少个消息
channel.basicQos(1);
// 5. 创建消费者(接收消息并处理消息);
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为:" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为:" + envelope.getExchange());
//消息id
System.out.println("消息id为:" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("消费者1-----接收到的消息为:" + new String(body, "utf-8"));
try {
Thread.sleep(1000);
//确认消息
/*
参数1:消息id
参数2:是否确认,false表示只有当前这条消息被处理
*/
channel.basicAck(envelope.getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
// 6. 监听队列 (需要持续监听队列消息,所以不要关闭资源)
/**
* 参数1:队列名
* 参数2:是否要自动确认;设置为true表示消息接收到自动向MQ回复接收到了,MQ则会将消息从队列中删除;
* 如果设置为false则需要手动确认
* 参数3:消费者
*/
channel.basicConsume(Producer.QUEUE_NAME,true,defaultConsumer);
//不关闭资源,应该一直监听消息
// channel.close();
// connection.close();
}
}
- 先启动消费者,再启动生产者 结果:发现一个消息只能被一个消费者接收,其它消费者是不能接收到同一条消息的
3、工作队列模式小结:
工作队列模式:一个消息只能被一个消费者接收,其它消费者是不能接收到同一条消息的。
应用场景:对于 任务过重或任务较多情况使用工作队列,可以提高任务处理的速度
添加对同一个队列的消费者来提高任务处理能力。