事件脉络消费处理逻辑修改

src/main/java/com/zzsn/event/kafka/KafkaConsumer.java

@@ -79,15 +79,23 @@ public class KafkaConsumer {
     public void eventContext(ConsumerRecord<String, String> record) {
         String value = record.value();
         if (StringUtils.isNotEmpty(value)) {
+            String subjectId;
+            Integer category = 2;
             try {
                 List<SubjectAnalysis> subjectAnalyses = JSON.parseArray(value, SubjectAnalysis.class);
-                for (SubjectAnalysis e : subjectAnalyses) {
-                    if (exist(e)) {
-                        e.setTitle(removeNonBmpUniCodes(e.getTitle()));
-                        subjectAnalysisService.save(e);
-                    }
+                subjectId = subjectAnalyses.get(0).getSubjectId();
+                subjectAnalyses.forEach(e -> {
+                    e.setCategory(category);
+                    e.setTitle(removeNonBmpUniCodes(e.getTitle()));
+                });
+                LambdaQueryWrapper<SubjectAnalysis> queryWrapper = Wrappers.lambdaQuery();
+                queryWrapper.eq(SubjectAnalysis::getSubjectId, subjectId).eq(SubjectAnalysis::getCategory, category);
+                int count = subjectAnalysisService.count(queryWrapper);
+                if (count > 0) {
+                    subjectAnalysisService.remove(queryWrapper);
                 }
-                log.info("id为-{}-的专题,此次-事件脉络-数据更新完成", subjectAnalyses.get(0).getSubjectId());
+                subjectAnalysisService.saveBatch(subjectAnalyses);
+                log.info("id为-{}-的专题,此次-事件脉络-数据更新完成", subjectId);
             } catch (Exception e) {
                 e.printStackTrace();
             }
@@ -103,7 +111,7 @@ public class KafkaConsumer {
     public void eventContext_fake(ConsumerRecord<String, String> record) {
         String value = record.value();
         if (StringUtils.isNotEmpty(value)) {
-            String subjectId = null;
+            String subjectId;
             Integer category = 3;
             try {
                 List<SubjectAnalysis> subjectAnalyses = JSON.parseArray(value, SubjectAnalysis.class);
@@ -119,10 +127,10 @@ public class KafkaConsumer {
                     subjectAnalysisService.remove(queryWrapper);
                 }
                 subjectAnalysisService.saveBatch(subjectAnalyses);
+                log.info("id为-{}-的专题,此次-伪事件脉络-数据更新完成", subjectId);
             } catch (Exception e) {
                 e.printStackTrace();
             }
-            log.info("id为-{}-的专题,此次-伪事件脉络-数据更新完成", subjectId);
         }
     }