rabbitmq之工作队列(一)

概念

工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。

1.轮训分发消息

所谓轮询分发消息，就是不考虑消费方的性能，采用发给a执行任务再发给b执行任务循环重复此动作。

1.1代码举例

/** * 创建一个获取信道的工具类 */publicclassRabbitMqUtils{publicstaticChannelgetChannel(){//创建一个连接工厂ConnectionFactoryfactory=newConnectionFactory();factory.setHost("192.168.136.128");//factory.setPort(15672);factory.setUsername("admin");factory.setPassword("admin");//创建连接Connectionconnection=null;Channelchannel=null;try{connection=factory.newConnection();//获取信道channel=connection.createChannel();}catch(IOExceptione){e.printStackTrace();}catch(TimeoutExceptione){e.printStackTrace();}returnchannel;}}

/** * 工作队列轮询分发测试 * 消费者1号 */publicclassConsumer01{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[]strings)throwsIOException{Channelchannel=RabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback=(consumerTag,delivery)->{Stringmessage=newString(delivery.getBody());System.out.println(message);};//取消消费的一个回调接口 如在消费的时候队列被删除掉了CancelCallbackcancelCallback=(consumerTag)->{System.out.println("消息消费被中断");};/** * 消费者消费消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答 * 3.消费者成功消费的回调 * 4.消息取消后的回调 */channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);}}

/** * 工作队列轮询分发测试 * 消费者2号 */publicclassConsumer02{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[]strings)throwsIOException{Channelchannel=RabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback=(consumerTag,delivery)->{Stringmessage=newString(delivery.getBody());System.out.println(message);};//取消消费的一个回调接口 如在消费的时候队列被删除掉了CancelCallbackcancelCallback=(consumerTag)->{System.out.println("消息消费被中断");};/** * 消费者消费消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答 * 3.消费者成功消费的回调 * 4.消息取消后的回调 */channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);}}

/** * 工作队列轮询分发测试 * 生产者 */publicclassProducer{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[]strings)throwsIOException{Channelchannel=RabbitMqUtils.getChannel();/** * 生成一个队列 * 1.队列名称 * 2.队列里面的消息是否持久化 默认消息存储在内存中 * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费 * 4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除 * 5.其他参数 */channel.queueDeclare(QUEUE_NAME,false,false,false,null);/** * 发送一个消息 * 1.发送到那个交换机 * 2.路由的 key 是哪个 * 3.其他的参数信息 * 4.发送消息的消息体 *///从控制台当中接受信息Scannerscanner=newScanner(System.in);while(scanner.hasNext()){Stringmessage=scanner.next();channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("发送消息完成:"+message);}}}

1.2观察结果

启动生产者发消息，启动消费者1和消费者2，观察结果

观察结果可以发现。生产者推送了4条消息，两个消费者都是各接受两条来处理的。

2.消息应答

消息应答就是为了保证消息在发送过程中不丢失引入的机制，当mq发送给消费者一条消息后，消费者便通知mq接到了该消息，此时mq就会根据消费者的确认应答删除该消息。

2.1自动消息应答

自动消息应答就是当消费者一但接受到消息就会立刻通知mq，mq会将该消息删除。

2.1.1 优缺点

• 缺点：会造成消息没有被完成处理即丢失消息的问题，当消费者接受到消息后，但在处理过程中消费者发生了异常，那么此时mq也已将消息删除，就会造成消失丢失的问题。

2.2手动消息应答

当消费者收到消息后不会立刻自动通知mq，而是可以在处理完成消息后的代码部分,人为加入basicAck()来通知mq。

2.2.1 优缺点

• 优点：提高消息处理的完整性，可以解决消费者接受到消息后在处理工程中出现异常，队列中的该消息不会被删除。消息自动重新入队

2.2.2 手动消息应答的方法

A.Channel.basicAck(用于肯定确认)
RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了
B.Channel.basicNack(用于否定确认) 根据requeue参数true:重回队列 false:丢弃消息
C.Channel.basicReject(用于否定确认) 消息重回队列

2.2.3 Multiple 的解释

Multiple 批量应答并且减少网络拥堵，但会可能会造成消息不会被完成处理。

2.3消息重新入队列

如果消费者由于某些原因失去连接，导致消息没有被ack确认应答，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

2.4手动消息应答代码测试

/** * 自动应答：消费者只要接收到消息就告诉生产者消息可以被删除了 * 测试： * 手动应答：在代码中使用basicAsk()通知生产者， * 测试效果：生产者一旦没有收到确认消息则会将改消息重新加入队列 */publicclassProducer{privatefinalstaticStringQUEUE_NAME="ack_queue";publicstaticvoidmain(String[]strings)throwsException{Channelchannel=RabbitMqUtils.getChannel();channel.queueDeclare(QUEUE_NAME,false,false,false,null);Scannerscanner=newScanner(System.in);System.out.println("请输入信息");while(scanner.hasNext()){Stringmessage=scanner.nextLine();channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));System.out.println("生产者发出消息"+message);}}}

/** * 消费者1号 */publicclassConsumer01{privatefinalstaticStringQUEUE_NAME="ack_queue";publicstaticvoidmain(String[]strings)throwsIOException{Channelchannel=RabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback=(consumerTag,delivery)->{Stringmessage=newString(delivery.getBody());System.out.println("Consumer01 接受到消息处理中...");SleepUtils.sleep(1);/** * 设置手动应答 * 1.消息标记 tag * 2.是否批量应答未应答消息 */channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);System.out.println("Consumer01 处理完成"+message);};//取消消费的一个回调接口 如在消费的时候队列被删除掉了CancelCallbackcancelCallback=(consumerTag)->{System.out.println("消息消费被中断");};/** * 消费者消费消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答 * 3.消费者成功消费的回调 * 4.消息取消后的回调 */channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);}}

publicclassConsumer02{privatefinalstaticStringQUEUE_NAME="ack_queue";publicstaticvoidmain(String[]strings)throwsIOException{Channelchannel=RabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback=(consumerTag,delivery)->{Stringmessage=newString(delivery.getBody());System.out.println("Consumer02 接受到消息处理中...");SleepUtils.sleep(30);/** * 设置手动应答 * 1.消息标记 tag * 2.是否批量应答未应答消息 */channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);System.out.println("Consumer02 处理完成"+message);};//取消消费的一个回调接口 如在消费的时候队列被删除掉了CancelCallbackcancelCallback=(consumerTag)->{System.out.println("消息消费被中断");};/** * 消费者消费消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答 * 3.消费者成功消费的回调 * 4.消息取消后的回调 */channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);}}

2.4.1观察结果

生产者发出同时发送 aa bb cc dd,mq默认轮询分发

c1接收到aa cc并完成

c2接收到 bb dd(ps:c2并不会因为还在处理bb就不会接收dd)
c2 处理完成bb 开始处理dd

c2处理dd时宕机

由于c2处理dd时宕机而没有确认应答，mq则会重新将dd加入到队列，重新送到可以处理的消费方进行处理。

3.持久化

默认情况下，mq是将队列和消息存入到缓存中的，因此当mq宕机那么就会使队列与消息丢失，因此可以通过mq持久化将队列与消息持久化到磁盘中，这样mq宕机后也不会丢失消息。

3.1队列持久化

在生产者声明队列时设置durable位true

/** * 生成一个队列 * 1.queue 队列名称 * 2.durable队列里面的消息是否持久化 默认消息存储在内存中 * 3.exclusive该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费 * 4.autoDelete是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除 * 5.arguments其他参数 *///将durable改为truechannel.queueDeclare(QUEUE_NAME,true,false,false,null);

但是需要注意的就是如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会出现错误。

以下为控制台中持久化与非持久化队列的 UI 显示区

3.2消息持久化

在生产者发布消息时设置props

4.不公平分发

不公平分发可以理解为能者多劳，可以根据消费方的性能进行多分发或少分发，这也做可以充分发挥高性能消费方的性能，减少低性能消费方的压力。因此在现实场景中一般使用不公平分发而不使用轮询分发。

//消费方设置prefetch = 1，即开启不公平分发channel.basicQos(1);

如上图c1与c2都开启了不公平分发，如果当前任务c1还没有处理完，那么mq就不会吧下一个任务分配给c1，下个任务而是将交给空闲中的其他消费者处理。如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的 worker 或者改变其他存储任务的策略。

5.预取值

预取值就是指，信道中可堆积的消息数量，只要消息没有堆积到设置的预取值数量，那么就不会因消费方的性能而不分发给该消费方。
举例：
c1处理时间平均1s设置预取值2，c2处理时间平均30s设置预取值5，此时生产者1秒内发送了7条数据数据，那么结果就是c1的信道中被分发了2条消息，c2分到了5条消息，后续的29s生产者连续发送消息，那么这些消息都会被c1处理掉，因为c2的预取值是5个而处理时间又是30s因此c2的信道堆积数量是满的，c1的信道每秒都会腾出一个位置接受新的消息。

//c1设置2个预取值intnum=2;channel.basicQos(num);

//c2设置5个预取值intnum=5;channel.basicQos(num);

6.消息确认

https://blog.csdn.net/qq_42872629/article/details/119985582?spm=1001.2014.3001.5501