Browse Source

实现本地消息表消息补偿,消息去重

feature-1.1
wuxicheng 3 years ago
parent
commit
f4edcac125
  1. 8
      bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/persist/RedisPersist.java
  2. 1
      bnyer-services/bnyer-order/src/main/java/com/bnyer/order/service/impl/OrderMqMessageRecordServiceImpl.java

8
bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/persist/RedisPersist.java

@ -30,11 +30,11 @@ public class RedisPersist implements IPersist {
} }
@Override @Override
public <T extends BaseMessage> boolean setConsumingIfNX(RepeatElement repeatElement, long dedupProcessingExpireMilliSeconds) { public <T extends BaseMessage> boolean setConsumingIfNX(RepeatElement repeatElement, long processingExpireMilliSeconds) {
String repeatKey = buildRepeatMessageRedisKey(repeatElement.getApplicationName(), repeatElement.getTopic(), repeatElement.getTag(), repeatElement.getMessageKey()); String repeatKey = buildRepeatMessageRedisKey(repeatElement.getApplicationName(), repeatElement.getTopic(), repeatElement.getTag(), repeatElement.getMessageKey());
//setnx, 成功就可以消费 //setnx, 成功就可以消费
Boolean execute = redisTemplate.execute((RedisCallback<Boolean>) redisConnection -> redisConnection.set(repeatKey.getBytes(), Boolean execute = redisTemplate.execute((RedisCallback<Boolean>) redisConnection -> redisConnection.set(repeatKey.getBytes(),
(CONSUME_STATUS_CONSUMING).getBytes(), Expiration.milliseconds(dedupProcessingExpireMilliSeconds), RedisStringCommands.SetOption.SET_IF_ABSENT)); (CONSUME_STATUS_CONSUMING).getBytes(), Expiration.milliseconds(processingExpireMilliSeconds), RedisStringCommands.SetOption.SET_IF_ABSENT));
if (execute == null) { if (execute == null) {
return false; return false;
} }
@ -48,9 +48,9 @@ public class RedisPersist implements IPersist {
} }
@Override @Override
public <T extends BaseMessage> void markConsumed(RepeatElement repeatElement, long dedupRecordReserveMinutes) { public <T extends BaseMessage> void markConsumed(RepeatElement repeatElement, long recordReserveMinutes) {
String repeatKey = buildRepeatMessageRedisKey(repeatElement.getApplicationName(), repeatElement.getTopic(), repeatElement.getTag(), repeatElement.getMessageKey()); String repeatKey = buildRepeatMessageRedisKey(repeatElement.getApplicationName(), repeatElement.getTopic(), repeatElement.getTag(), repeatElement.getMessageKey());
redisTemplate.opsForValue().set(repeatKey, CONSUME_STATUS_SUCCESS, dedupRecordReserveMinutes, TimeUnit.MINUTES); redisTemplate.opsForValue().set(repeatKey, CONSUME_STATUS_SUCCESS, recordReserveMinutes, TimeUnit.MINUTES);
} }
@Override @Override

1
bnyer-services/bnyer-order/src/main/java/com/bnyer/order/service/impl/OrderMqMessageRecordServiceImpl.java

@ -116,6 +116,7 @@ public class OrderMqMessageRecordServiceImpl implements OrderMqMessageRecordServ
orderMqMessageRecord.setTag(tag); orderMqMessageRecord.setTag(tag);
orderMqMessageRecord.setCreateTime(new Date()); orderMqMessageRecord.setCreateTime(new Date());
orderMqMessageRecord.setMessageKey(IdUtils.randomUUID()); orderMqMessageRecord.setMessageKey(IdUtils.randomUUID());
orderMqMessageRecord.setContent(JSON.toJSONString(message));
orderMqMessageRecordMapper.insert(orderMqMessageRecord); orderMqMessageRecordMapper.insert(orderMqMessageRecord);
return orderMqMessageRecord; return orderMqMessageRecord;
} }

Loading…
Cancel
Save