甘肃省建设工程网上投标网站,好玩的网游,网站建设的安全防护方法,智库门户网站建设在Java中#xff0c;可以使用消息队列来实现消息的异步处理#xff0c;其中常用的消息队列有 RabbitMQ、ActiveMQ、Kafka 等。
什么是幂等性#xff1f;
幂等性是指无论操作执行多少次#xff0c;都是得到相同的结果#xff0c;而不会产生其他副作用。
在rabbitMQ中
什…在Java中可以使用消息队列来实现消息的异步处理其中常用的消息队列有 RabbitMQ、ActiveMQ、Kafka 等。
什么是幂等性
幂等性是指无论操作执行多少次都是得到相同的结果而不会产生其他副作用。
在rabbitMQ中
什么是消息重复消费
同一条消息在MQ中被消费多次
出现重复消费的原因
生产者发送一条消息到rabbitMQ但rabbitMQ尚未收到消费者的确认会认为消息消费未被消费而重新发送。
网络不稳定、消费者故障、网络分区、消息重复传递策略、消费者超时设置不当
为什么需要避免重复消费
业务错误我本来写的业务逻辑就是只要执行一次
数据重复数据插入重复破坏数据唯一性
资源浪费占用系统资源降低系统性能
如何避免消息重复消费
消息去重
通过记录已经消费过的消息在消息到达时检查它是否已经在记录中存在从而避免重复处理。 if (!processedMessages.contains(message)) {processMessage(message);processedMessages.add(message);}
消息幂等性
分布式锁(消息幂等性)
使用UUID生成唯一Id 作为messageId
使用了唯一的消息ID来确保同一条消息只会被处理一次。
AMQP.BasicProperties properties new AMQP.BasicProperties().builder().messageId(UUID.randomUUID().toString()) // 唯一标识.build();if (!isMessageProcessed(messageId)) {processMessage(message);saveProcessedMessage(messageId);}
消费者先查询该消息是否已经被处理过如果没有被处理过则调用processMessage()方法处理该消息并使用 saveProcessedMessage()方法保存已经处理过的消息。 //手动ackchannel.basicAck(message.getMessageProperties().getDeliveryTag(),false在处理完消息后还需要调用channel.basicAck(envelope.getDeliveryTag(), false)方法确认消息已经被消费。这是因为RabbitMQ是一个消息的投递机制只有在消费者确认了消息已经被处理后才会从消息队列中删除该消息。
使用redis实现避免重复消费
生产者
AMQP.BasicProperties properties new AMQP.BasicProperties().builder().deliveryMode(1) // 指定消息是否需要持久化 1-需要 2-不需要.messageId(UUID.randomUUID().toString()) // 唯一标识.build();消费者 String result jedis.set(messageId, 0, NX, EX, 10);if (result ! null result.equalsIgnoreCase(OK)){System.out.println(接收到消息 new String(body,UTF-8));//消费成功 set messageId - 1jedis.set(messageId,1);channel.basicAck(envelope.getDeliveryTag(),false);}else {//如果1中的setnx失败获取key对应的value如果是1设置ack 如果是0 returnString s jedis.get(messageId);if (1.equalsIgnoreCase(s)){//消费完了channel.basicAck(envelope.getDeliveryTag(),false);}}spring-boot
如果存在设置value为1如果value是1ack
事务性消费
消费状态追踪
