🍁博客主页:👉不会压弯的小飞侠 ?欢迎关注:👉点赞👍收藏?留言? ?系列专栏:👉Linux专栏 🔥欢迎大佬指正,一起学习!一起加油!
🍁模式说明
-
工作队列(又名:任务队列)背后的主要思想是避免 立即执行资源密集型任务,必须等待 它要完成。相反,我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。正在运行的工作进程 在后台将弹出任务并最终执行 工作。当您运行许多工作线程时,任务将在它们之间共享。 -
与入门程序的简单模式相比,多了一个或一些消费端,多个消费端共同消费同一个队列中的消息。 -
对于任务较多情况使用工作队列可以提高任务处理的速度。
🍁工作模式完成消息传递
- 在学习简单模式的时候工程已创建完成,只需要在不同包下测试即可。
- 需要创建一个工具类,将连接工厂,抽取出来。
- 在util包下创建ConnectionUtil类
public class ConnectionUtil {
public static Connection getConnection() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("**************");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("admin");
factory.setPassword("root");
Connection connection = factory.newConnection();
return connection;
}
public static void main(String[] args) throws Exception {
Connection con = ConnectionUtil.getConnection();
System.out.println(con);
con.close();
}
}
public class Producer {
static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
for (int i = 1; i <= 10; i++) {
String body = i+"Hello RabbitMQ!!!";
channel.basicPublish("",QUEUE_NAME,null,body.getBytes());
}
channel.close();
connection.close();
}
}
public class Consumer1 {
static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, true, 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("body:" + new String(body));
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
public class Consumer2 {
static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, true, 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("body:" + new String(body));
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
- 测试
- 启动两个消费者,然后再启动生产者发送消息;到IDEA的两个消费者对应的控制台查看是否竞争性的接收到消息。
🍁总结
- 在一个队列中如果有多个消费者,那么消费者之间对于同一个消息的关系是竞争的关系
- Work Queues 对于任务较多情况使用工作队列可以提高任务处理的速度。
- 一个生产者、多个消费者(竞争关系),不需要设置交换机(使用默认的交换机)
|