深圳网站建设迈,flash 做ppt的模板下载网站有哪些,PHP与网站建设的课后笔记,奉贤庄行网站建设在 RabbitMQ 中#xff0c;消息消费者对消息的签收#xff08;acknowledgment#xff09;可以通过三种方式进行管理#xff1a;自动签收、手动签收 和 拒绝签收。它们主要控制消费者如何处理消息确认和消息的重新排队。下面详细讲解它们的区别#xff0c;并通过代码示例展…在 RabbitMQ 中消息消费者对消息的签收acknowledgment可以通过三种方式进行管理自动签收、手动签收 和 拒绝签收。它们主要控制消费者如何处理消息确认和消息的重新排队。下面详细讲解它们的区别并通过代码示例展示。
1. 自动签收Auto Acknowledgment
在自动签收模式下RabbitMQ 会在消息被传递到消费者时自动进行消息确认acknowledge无需消费者明确地通知 RabbitMQ 消息已经被成功处理。这意味着只要消费者接受到消息它就认为消息已成功处理并确认。
优点 消费者的代码较为简单不需要手动确认。 缺点 如果消费者处理消息时发生错误RabbitMQ 已经认为消息被成功处理可能会导致消息丢失或无法重新投递。
2. 手动签收Manual Acknowledgment
在手动签收模式下消费者需要显式地通知 RabbitMQ 消息已经被成功处理。消费者通过调用 channel.basicAck() 来手动确认消息。这种模式提供了更大的控制可以确保只有在消息成功处理后才确认消息。
优点 可以确保消息在处理成功后才会被确认失败时可以拒绝签收并重新投递。 缺点 需要开发者手动管理确认过程代码相对复杂。
3. 拒绝签收Reject Acknowledgment
拒绝签收指的是消费者告知 RabbitMQ 它无法处理某个消息可以通过 channel.basicReject() 或 channel.basicNack() 来拒绝该消息。拒绝签收的消息可以被重新投递到队列中或者直接丢弃具体取决于设置的参数。
优点 可以明确告诉 RabbitMQ 消息无法处理并且可以选择将消息重新排队供其他消费者处理。 缺点 如果消息被拒绝并重新排队可能会导致消息的重复消费。
代码示例
下面展示了 自动签收、手动签收 和 拒绝签收 的代码示例。
1. 自动签收Auto Acknowledgment
package com.home.consumer;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class AutoAckConsumer {// 自动签收RabbitListener(queues myQueue, ackMode AUTO)public void handleMessage(String msg) {System.out.println(Received message (Auto Acknowledgment): msg);// 处理完消息后自动确认}
}在自动签收模式下消息会在 handleMessage 方法执行完毕后自动被 RabbitMQ 确认。
2. 手动签收Manual Acknowledgment
package com.home.consumer;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.MessageListener;
import org.springframework.stereotype.Component;Component
public class ManualAckConsumer {// 手动签收RabbitListener(queues myQueue, ackMode MANUAL)public void handleMessage(String msg, Message message) {System.out.println(Received message (Manual Acknowledgment): msg);// 处理完消息后显式调用 channel.basicAck() 来确认try {// 消息处理逻辑// 如果处理成功手动确认消息message.getMessageProperties().getChannel().basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 处理失败时拒绝签收消息message.getMessageProperties().getChannel().basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}
}在手动签收模式下您需要在代码中手动调用 basicAck() 来确认消息或者在出现错误时使用 basicNack() 或 basicReject() 来拒绝消息。
3. 拒绝签收Reject Acknowledgment
package com.home.consumer;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.MessageListener;
import org.springframework.stereotype.Component;Component
public class RejectAckConsumer {// 拒绝签收RabbitListener(queues myQueue, ackMode MANUAL)public void handleMessage(String msg, Message message) {System.out.println(Received message (Reject Acknowledgment): msg);// 如果消息处理失败则拒绝签收if (msg.contains(error)) {try {// 拒绝消息并重新排队message.getMessageProperties().getChannel().basicReject(message.getMessageProperties().getDeliveryTag(), true);} catch (Exception e) {System.out.println(Failed to reject the message.);}} else {// 处理成功时确认消息try {message.getMessageProperties().getChannel().basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {System.out.println(Failed to acknowledge the message.);}}}
}在拒绝签收模式下当消费者无法处理某个消息时可以调用 basicReject() 或 basicNack() 来拒绝该消息并根据需要选择是否将其重新排队。如果选择不重新排队消息将会丢失。
总结
签收模式描述优点缺点自动签收消费者收到消息后自动确认代码简洁、开发快速消息处理失败时可能丢失消息手动签收消费者需要显式确认消息处理成功或失败使用 basicAck可确保消息成功处理后才确认失败时可重试或重新排队需要开发者手动管理确认过程代码复杂拒绝签收消费者拒绝消息并可选择重新排队或丢弃可以显式地告诉 RabbitMQ 消息无法处理且重新排队如果不设置合适的重试机制可能导致消息重复消费
在实际使用时选择何种签收方式依赖于您的应用需求通常 手动签收 会提供更好的控制尤其是在高可靠性需求的场景下。
拒绝签收后的操作
在 RabbitMQ 中当你使用拒绝签收basicReject() 或 basicNack()时可以选择是否将消息重新排队。你可以通过设置相应的参数来决定消息的处理方式。
basicReject() 和 basicNack() 方法的参数
这两个方法允许你传递一个参数指示是否要重新排队消息
requeue 参数如果为 true消息将被重新排队等待其他消费者消费如果为 false消息将不会重新排队可能会丢失取决于配置。
basicReject()
channel.basicReject(deliveryTag, requeue);
deliveryTag是消息的唯一标识符指示哪条消息被拒绝。requeue布尔值指示消息是否应该重新排队。如果为 true消息将重新排队到队列中如果为 false消息将丢失或直接被丢弃。
basicNack()
channel.basicNack(deliveryTag, multiple, requeue);
deliveryTag与 basicReject() 中相同用于标识消息。multiple是否拒绝所有比当前 deliveryTag 小的消息。如果为 true则拒绝多个消息如果为 false则仅拒绝当前消息。requeue与 basicReject() 中相同控制是否将消息重新排队。
1. 使用 basicReject() 来拒绝并重新排队
package com.home.consumer;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class RejectAndRequeueConsumer {RabbitListener(queues myQueue, ackMode MANUAL)public void handleMessage(String msg, Message message) {System.out.println(Received message (Reject and Requeue): msg);// 如果消息处理失败拒绝消息并重新排队if (msg.contains(error)) {try {// 拒绝并重新排队message.getMessageProperties().getChannel().basicReject(message.getMessageProperties().getDeliveryTag(), true);System.out.println(Message rejected and requeued);} catch (Exception e) {System.out.println(Failed to reject and requeue the message.);}} else {// 处理成功时确认消息try {message.getMessageProperties().getChannel().basicAck(message.getMessageProperties().getDeliveryTag(), false);System.out.println(Message processed successfully and acknowledged);} catch (Exception e) {System.out.println(Failed to acknowledge the message.);}}}
}打印接收到的消息首先输出消息内容 msg。 消息失败处理如果消息内容包含 error通过 msg.contains(error) 检查认为处理失败执行以下操作 拒绝并重新排队通过 basicReject 方法拒绝消息且设置 requeue true即将消息重新放回队列等待其他消费者处理。此时RabbitMQ 会重新将消息排入队列等待下一次消费。异常处理如果拒绝和重新排队消息过程中出现异常捕获异常并打印错误信息。 消息成功处理如果消息内容没有错误即不包含 error则认为消息处理成功执行以下操作 确认消息通过 basicAck 方法手动确认消息已成功处理。异常处理如果在确认过程中发生异常捕获并打印错误信息。
2. 使用 basicReject() 来拒绝并丢弃消息
package com.home.consumer;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class RejectAndDiscardConsumer {RabbitListener(queues myQueue, ackMode MANUAL)public void handleMessage(String msg, Message message) {System.out.println(Received message (Reject and Discard): msg);// 如果消息处理失败拒绝消息并丢弃if (msg.contains(error)) {try {// 拒绝并丢弃消息不重新排队message.getMessageProperties().getChannel().basicReject(message.getMessageProperties().getDeliveryTag(), false);System.out.println(Message rejected and discarded);} catch (Exception e) {System.out.println(Failed to reject and discard the message.);}} else {// 处理成功时确认消息try {message.getMessageProperties().getChannel().basicAck(message.getMessageProperties().getDeliveryTag(), false);System.out.println(Message processed successfully and acknowledged);} catch (Exception e) {System.out.println(Failed to acknowledge the message.);}}}
}打印接收到的消息首先输出接收到的消息内容 msg。 消息失败处理如果消息的内容包含 error通过 msg.contains(error) 判断就认为消息处理失败并执行以下操作 拒绝并丢弃消息通过 basicReject 方法拒绝消息并且将 requeue 设置为 false即消息不会重新排回队列而是直接丢弃。这样该消息就不会再被其他消费者处理。异常处理如果拒绝和丢弃消息时发生异常则捕获异常并打印错误信息。 消息成功处理如果消息内容不包含 error认为消息处理成功执行以下操作 确认消息通过 basicAck 方法手动确认消息已被成功处理。异常处理如果确认消息时发生异常则捕获并打印错误信息。
3.使用 basicNack() 来拒绝多个消息并重新排队
package com.home.consumer;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class NackAndRequeueConsumer {RabbitListener(queues myQueue, ackMode MANUAL)public void handleMessage(String msg, Message message) {System.out.println(Received message (Nack and Requeue): msg);// 如果消息处理失败拒绝多个消息并重新排队if (msg.contains(error)) {try {// 拒绝多个消息并重新排队message.getMessageProperties().getChannel().basicNack(message.getMessageProperties().getDeliveryTag(), false, true);System.out.println(Message Nacked and requeued);} catch (Exception e) {System.out.println(Failed to nack and requeue the message.);}} else {// 处理成功时确认消息try {message.getMessageProperties().getChannel().basicAck(message.getMessageProperties().getDeliveryTag(), false);System.out.println(Message processed successfully and acknowledged);} catch (Exception e) {System.out.println(Failed to acknowledge the message.);}}}
}打印接收到的消息首先输出接收到的消息内容 msg。 消息失败处理如果消息内容包含 error通过 msg.contains(error) 判断表示消息处理失败执行以下操作 拒绝多个消息并重新排队通过 basicNack 方法拒绝消息并设置以下参数 multiple false表示只拒绝当前消息。如果设置为 true则会拒绝当前消息以及所有较早的消息。requeue true表示将消息重新排回队列等待其他消费者处理。 异常处理如果在执行 basicNack 时出现异常捕获异常并打印错误信息。 消息成功处理如果消息内容不包含 error则认为消息处理成功执行以下操作 确认消息通过 basicAck 方法手动确认消息已经成功处理。异常处理如果在确认消息时发生异常捕获并打印错误信息。
basicNack 和 basicAck 的区别 basicNack这个方法是用于拒绝消息与 basicReject 类似。basicNack 还可以处理多个消息并允许你选择是否重新排队。与 basicReject 不同basicNack 能够处理批量拒绝消息。 multiple false表示仅拒绝当前消息。requeue true表示将消息重新排回队列。 basicAck这是用来确认消息已经成功处理的。当消息被成功消费后消费者需要调用该方法来告诉 RabbitMQ 消息已经被处理RabbitMQ 可以删除该消息。
