新增redis锁解决多消费者消费事件数据重新问题

jeecg-boot-parent/jeecg-boot-module-property/src/main/java/org/jeecg/pm/mq/EventMessageListener.java

@@ -67,14 +67,23 @@ public class EventMessageListener {
 
 
     @JmsListener(destination = EPS_TOPIC)
     @JmsListener(destination = EPS_TOPIC)
     public void epsProcessMessage(BytesMessage bytesMessage) throws Exception {
     public void epsProcessMessage(BytesMessage bytesMessage) throws Exception {
+        String lock = EPS_TOPIC;
+        try {
             final byte[] bytes = new byte[(int) bytesMessage.getBodyLength()];
             final byte[] bytes = new byte[(int) bytesMessage.getBodyLength()];
             bytesMessage.readBytes(bytes);
             bytesMessage.readBytes(bytes);
+            // 壳文件字段，EventDis类为event_dis.proto文件解析而来，CommEventLog类为事件壳文件类
             final EventDis.CommEventLog commEventLog = EventDis.CommEventLog.parseFrom(bytes);
             final EventDis.CommEventLog commEventLog = EventDis.CommEventLog.parseFrom(bytes);
+            lock = lock + commEventLog.getLogId();
+            if (redisTemplate.opsForValue().setIfAbsent(lock, lock)) {
                 final PmEventInfo pmEventInfo = this.recode(commEventLog);
                 final PmEventInfo pmEventInfo = this.recode(commEventLog);
                 pmEventInfo.setSubsystem(PmEventInfo.EPS);
                 pmEventInfo.setSubsystem(PmEventInfo.EPS);
                 pmEventInfoService.saveEvent(pmEventInfo);
                 pmEventInfoService.saveEvent(pmEventInfo);
                 log.info("EPS_TOPIC:{}", pmEventInfo);
                 log.info("EPS_TOPIC:{}", pmEventInfo);
             }
             }
+        } finally {
+            redisTemplate.delete(lock);
+        }
+    }
 
 
 
 
     @SneakyThrows
     @SneakyThrows