工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务,而不得不等待它完成。 相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进 程将弹出任务并最终执行作业。当有多个工作线程时,这些工作线程将一起处理这些任务。
一、 轮训分发消息
在这个案例中我们会启动两个工作线程,一个消息发送线程,我们来看看他们两个工作线程 是如何工作的
1. 抽取工具类
package com.mangoubiubiu.utils; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class RabbitMqUtils { //得到一个连接的 channel public static Channel getChannel() throws Exception{ //创建一个连接工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.117.134"); factory.setUsername("admin"); factory.setPassword("123"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); return channel; } }
2. 启动两个工作线程
package com.mangoubiubiu.two; import com.mangoubiubiu.utils.RabbitMqUtils; import com.rabbitmq.client.CancelCallback; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; public class Work01 { private static final String QUEUE_NAME="hello1"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //消息的接收 DeliverCallback deliverCallback=(consumerTag, delivery)->{ String receivedMessage = new String(delivery.getBody()); System.out.println("接收到消息:"+receivedMessage); }; //消息接收被取消时 执行下面的内容 CancelCallback cancelCallback=(consumerTag)->{ System.out.println(consumerTag+"消费者取消消费接口回调逻辑"); }; System.out.println("C2 消费者启动等待消费......"); channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback); } }
3. 启动一个发送线程
package com.mangoubiubiu.two; import com.mangoubiubiu.utils.RabbitMqUtils; import com.rabbitmq.client.Channel; import java.util.Scanner; public class Task01 { private static final String QUEUE_NAME="hello1"; public static void main(String[] args) throws Exception { try(Channel channel= RabbitMqUtils.getChannel();) { /** * 入参说明 * 1 队列名称 * 2 队列里面的消息是否持久化(磁盘) 默认情况消息存储在内存中 * 3 该队列是否只供一个消费者进行消费 是否进行消息共享,true 可以多个消费者消费 false 只能一个消费者消费 * 4 是否自动删除 最后一个消费者断开连接以后 该队一句是否自动删除 true自动删除 false不自动删除 * 5 其他参数 */ channel.queueDeclare(QUEUE_NAME,false,false,false,null); //从控制台当中接受信息 Scanner scanner = new Scanner(System.in); while (scanner.hasNext()){ String message = scanner.next(); /** * 入参说明 * 1 发送到哪个交换机 * 2 路由的Key值是哪个 本次是队列的名称 * 3 其它参数信息 * 4 发送消息的消息体 */ channel.basicPublish("",QUEUE_NAME,null,message.getBytes()); System.out.println("发送消息完成:"+message); } } } }
4. 结果展示
通过程序执行发现生产者总共发送 4 个消息,消费者 1 和消费者 2 分别分得两个消息,并且 是按照有序的一个接收一次消息
标签:队列,Queues,Work,RabbitMQ,线程,rabbitmq,import,com,channel From: https://www.cnblogs.com/mangoubiubiu/p/17009185.html