文章目录概念1.轮训分发消息1.1代码举例1.2观察结果2.消息应答2.1自动消息应答2.1.1 优缺点2.2手动消息应答2.2.1 优缺点2.2.2 手动消息应答的方法2.2.3 Multiple 的解释2.3消息重新入队列2.4手动消息应答代码测试2.4.1观察结果3.持久化3.1队列持久化3.2消息持久化4.不公平分发5.预取值6.消息确认概念工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时这些工作线程将一起处理这些任务。1.轮训分发消息所谓轮询分发消息就是不考虑消费方的性能采用发给a执行任务再发给b执行任务循环重复此动作。1.1代码举例/** * 创建一个获取信道的工具类 */publicclassRabbitMqUtils{publicstaticChannelgetChannel(){//创建一个连接工厂ConnectionFactoryfactorynewConnectionFactory();factory.setHost(192.168.136.128);//factory.setPort(15672);factory.setUsername(admin);factory.setPassword(admin);//创建连接Connectionconnectionnull;Channelchannelnull;try{connectionfactory.newConnection();//获取信道channelconnection.createChannel();}catch(IOExceptione){e.printStackTrace();}catch(TimeoutExceptione){e.printStackTrace();}returnchannel;}}/** * 工作队列轮询分发测试 * 消费者1号 */publicclassConsumer01{privatefinalstaticStringQUEUE_NAMEhello;publicstaticvoidmain(String[]strings)throwsIOException{ChannelchannelRabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback(consumerTag,delivery)-{StringmessagenewString(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_NAMEhello;publicstaticvoidmain(String[]strings)throwsIOException{ChannelchannelRabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback(consumerTag,delivery)-{StringmessagenewString(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_NAMEhello;publicstaticvoidmain(String[]strings)throwsIOException{ChannelchannelRabbitMqUtils.getChannel();/** * 生成一个队列 * 1.队列名称 * 2.队列里面的消息是否持久化 默认消息存储在内存中 * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费 * 4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除 * 5.其他参数 */channel.queueDeclare(QUEUE_NAME,false,false,false,null);/** * 发送一个消息 * 1.发送到那个交换机 * 2.路由的 key 是哪个 * 3.其他的参数信息 * 4.发送消息的消息体 *///从控制台当中接受信息ScannerscannernewScanner(System.in);while(scanner.hasNext()){Stringmessagescanner.next();channel.basicPublish(,QUEUE_NAME,null,message.getBytes());System.out.println(发送消息完成:message);}}}1.2观察结果启动生产者发消息启动消费者1和消费者2观察结果观察结果可以发现。生产者推送了4条消息两个消费者都是各接受两条来处理的。2.消息应答消息应答就是为了保证消息在发送过程中不丢失引入的机制当mq发送给消费者一条消息后消费者便通知mq接到了该消息此时mq就会根据消费者的确认应答删除该消息。2.1自动消息应答自动消息应答就是当消费者一但接受到消息就会立刻通知mqmq会将该消息删除。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_NAMEack_queue;publicstaticvoidmain(String[]strings)throwsException{ChannelchannelRabbitMqUtils.getChannel();channel.queueDeclare(QUEUE_NAME,false,false,false,null);ScannerscannernewScanner(System.in);System.out.println(请输入信息);while(scanner.hasNext()){Stringmessagescanner.nextLine();channel.basicPublish(,QUEUE_NAME,null,message.getBytes(UTF-8));System.out.println(生产者发出消息message);}}}/** * 消费者1号 */publicclassConsumer01{privatefinalstaticStringQUEUE_NAMEack_queue;publicstaticvoidmain(String[]strings)throwsIOException{ChannelchannelRabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback(consumerTag,delivery)-{StringmessagenewString(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_NAMEack_queue;publicstaticvoidmain(String[]strings)throwsIOException{ChannelchannelRabbitMqUtils.getChannel();//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback(consumerTag,delivery)-{StringmessagenewString(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 开始处理ddc2处理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消息持久化在生产者发布消息时设置props4.不公平分发不公平分发可以理解为能者多劳可以根据消费方的性能进行多分发或少分发这也做可以充分发挥高性能消费方的性能减少低性能消费方的压力。因此在现实场景中一般使用不公平分发而不使用轮询分发。//消费方设置prefetch 1即开启不公平分发channel.basicQos(1);如上图c1与c2都开启了不公平分发如果当前任务c1还没有处理完那么mq就不会吧下一个任务分配给c1下个任务而是将交给空闲中的其他消费者处理。如果所有的消费者都没有完成手上任务队列还在不停的添加新任务队列有可能就会遇到队列被撑满的情况这个时候就只能添加新的 worker 或者改变其他存储任务的策略。5.预取值预取值就是指信道中可堆积的消息数量只要消息没有堆积到设置的预取值数量那么就不会因消费方的性能而不分发给该消费方。举例c1处理时间平均1s设置预取值2c2处理时间平均30s设置预取值5此时生产者1秒内发送了7条数据数据那么结果就是c1的信道中被分发了2条消息c2分到了5条消息后续的29s生产者连续发送消息那么这些消息都会被c1处理掉因为c2的预取值是5个而处理时间又是30s因此c2的信道堆积数量是满的c1的信道每秒都会腾出一个位置接受新的消息。//c1设置2个预取值intnum2;channel.basicQos(num);//c2设置5个预取值intnum5;channel.basicQos(num);6.消息确认https://blog.csdn.net/qq_42872629/article/details/119985582?spm1001.2014.3001.5501