剔除旧版本

src/main/java/com/winhc/config/ConfigConstant.java

@@ -1,17 +1,8 @@
 package com.winhc.config;
 
-import lombok.AllArgsConstructor;
-import org.neo4j.driver.AuthTokens;
-import org.neo4j.driver.Config;
-import org.neo4j.driver.Driver;
-import org.neo4j.driver.GraphDatabase;
+
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
-
-import java.util.concurrent.TimeUnit;
 
 @Configuration
 public class ConfigConstant {
@@ -19,6 +10,8 @@ public class ConfigConstant {
     @Value("${spring.kafka.topic_case}")
     public String topic_case;
 
+    @Value("${spring.kafka.topic_case_v2}")
+    public String topic_case_v2;
 
 
 }

src/main/java/com/winhc/kafka/consumer/KafkaConsumerCaseMerge.java

@@ -21,12 +21,13 @@ import java.util.Map;
 
 /**
  * @author π
- * @Description:案件增量消费
+ * @Description:案件增量消费(旧版本取消)
  * @date 2021/5/17 16:12
  */
 @Slf4j
-@Service
+//@Service
 @AllArgsConstructor
+@Deprecated
 public class KafkaConsumerCaseMerge {
 
     private final Map<String, RelationService> map;

src/main/java/com/winhc/kafka/consumer/KafkaConsumerCaseMergeV2.java

@@ -0,0 +1,61 @@
+package com.winhc.kafka.consumer;
+
+import com.winhc.common.enums.CompanyEnum;
+import com.winhc.config.ConfigConstant;
+import com.winhc.kafka.KafkaProduce;
+import com.winhc.service.RelationService;
+import com.winhc.utils.CompanyUtils;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author π
+ * @Description:案件增量消费v2
+ * @date 2021/8/30 16:12
+ */
+@Slf4j
+@Service
+@AllArgsConstructor
+public class KafkaConsumerCaseMergeV2 {
+
+    private final Map<String, RelationService> map;
+    @Autowired
+    KafkaProduce kafkaProduce;
+    @Autowired
+    ConfigConstant configConstant;
+
+    @KafkaListener(id = "${spring.kafka.topic_case_v2}"
+            , topics = "${spring.kafka.topic_case_v2}"
+            , groupId = "${spring.kafka.consumer.group-id}", containerFactory = "containerFactory", errorHandler = "KafkaConsumerCaseMergeV2")
+    public void consumerCompanyNode(List<ConsumerRecord<?, ?>> records) {
+        List<Map<String, Object>> listMap = CompanyUtils.map(records);
+        this.map.get(CompanyEnum.TopicType.CASE_RELATION_INCR.VALUE).save(CompanyUtils.filterList(listMap, CompanyEnum.TopicType.CASE_RELATION_INCR.CODE));
+        this.map.get(CompanyEnum.TopicType.CASE_NODE_INCR.VALUE).save(CompanyUtils.filterList(listMap, CompanyEnum.TopicType.CASE_NODE_INCR.CODE));
+        this.map.get(CompanyEnum.TopicType.CASE_RELATION_UPDATE.VALUE).save(CompanyUtils.filterList(listMap, CompanyEnum.TopicType.CASE_RELATION_UPDATE.CODE));
+        this.map.get(CompanyEnum.TopicType.CASE_NODE_UPDATE.VALUE).save(CompanyUtils.filterList(listMap, CompanyEnum.TopicType.CASE_NODE_UPDATE.CODE));
+    }
+
+    /**
+     * 因为手动确认消费，若消费失败，记录重刷
+     */
+    @Bean("KafkaConsumerCaseMergeV2")
+    public ConsumerAwareListenerErrorHandler dealError() {
+        return (message, e, consumer) -> {
+            List<String> list = CompanyUtils.toMessage((List<ConsumerRecord>) message.getPayload());
+            for (String msg : list) {
+                kafkaProduce.produce(configConstant.topic_case, msg);
+            }
+            log.error("KafkaConsumerCaseMerge error: {}", e.toString());
+            return null;
+        };
+    }
+}

src/main/java/com/winhc/kafka/consumer/KafkaConsumerNeo4j.java

@@ -23,6 +23,7 @@ import java.util.Map;
 @Slf4j
 //@Service
 @AllArgsConstructor
+@Deprecated
 public class KafkaConsumerNeo4j {
 
     private final Map<String, RelationService> map;

src/main/java/com/winhc/kafka/consumer/KafkaConsumerPersonMergeV2.java

@@ -24,6 +24,7 @@ import java.util.Map;
 @Slf4j
 //@Service
 @AllArgsConstructor
+@Deprecated
 public class KafkaConsumerPersonMergeV2 {
 
     private final Map<String, RelationService> map;

src/main/java/com/winhc/service/impl/CaseNodeIncrImpl.java

@@ -35,8 +35,9 @@ public class CaseNodeIncrImpl implements RelationService {
         Session session = driver.session();
         final String cql = "\nWITH  {batch_list} AS batch_list \n" +
                 "UNWIND batch_list AS row \n" +
-                " MERGE(s:CASE{case_id:row.start_id})\n" +
-                " SET s.case_id = row.start_id\n" ;
+                " MERGE(s:CASE{case_id:row.start_id})\n"
+//                " SET s.case_id = row.start_id\n"
+                ;
 //                " WITH s\n" +
 //                " CALL apoc.path.subgraphNodes(s, {maxLevel:-1}) YIELD node\n" +
 //                " WITH node,s\n" +

src/main/java/com/winhc/service/impl/CaseRelationIncrImpl.java

@@ -36,9 +36,9 @@ public class CaseRelationIncrImpl implements RelationService {
         final String cql = "\nWITH  {batch_list} AS batch_list \n" +
                 "UNWIND batch_list AS row \n" +
                 " MERGE(s:CASE{case_id:row.start_id})\n" +
-                " SET s.case_id = row.start_id\n" +
+//                " SET s.case_id = row.start_id\n" +
                 " MERGE(m:CASE{case_id:row.end_id})\n" +
-                " SET m.case_id = row.end_id\n" +
+//                " SET m.case_id = row.end_id\n" +
                 " MERGE(s)-[r:RELATION]->(m)\n" +
                 " SET r.connect_type=row.connect_type\n" ;
 //                " WITH s,m\n" +

src/main/resources/application-dev.properties

@@ -45,6 +45,7 @@ spring.kafka.topic_node_relation_union=inc_node_relation_union_dev
 spring.kafka.topic_person_companys=inc_person_companys_dev
 spring.kafka.topic_person_merge=inc_person_merge_dev
 spring.kafka.topic_case=inc_judicial_case_union_dev
+spring.kafka.topic_case_v2=inc_judicial_case_union_v2_dev
 spring.kafka.topic_person_merge_v2=inc_person_merge_v2_dev
 #spring.kafka.bootstrap-servers=192.168.4.237:9092,192.168.4.235:9092,192.168.4.236:9092
 #spring.kafka.topic=xf_test

src/main/resources/application-prd.properties

@@ -8,7 +8,8 @@ spring.data.neo4j.uri.v1=bolt://192.168.2.57:7687
 #Neo4j配置(第二台机器)
 spring.data.neo4j.username.v2=neo4j
 spring.data.neo4j.password.v2=neo4j168
-spring.data.neo4j.uri.v2=bolt://192.168.2.60:7687
+#spring.data.neo4j.uri.v2=bolt://192.168.2.60:7687
+spring.data.neo4j.uri.v2=bolt://192.168.2.62:7687
 
 
 #spring.datasource.url = jdbc:mysql://rm-uf61r3m23ba1p5z3dfo.mysql.rds.aliyuncs.com:3306/prism1?useUnicode=true&characterEncoding=utf-8
@@ -38,6 +39,7 @@ spring.kafka.topic_node_relation_union=inc_node_relation_union
 spring.kafka.topic_person_companys=inc_person_companys
 spring.kafka.topic_person_merge=inc_person_merge
 spring.kafka.topic_case=inc_judicial_case_union
+spring.kafka.topic_case_v2=inc_judicial_case_union_v2
 spring.kafka.topic_person_merge_v2=inc_person_merge_v2
 #=============== provider  =======================
 spring.kafka.producer.retries=3