update

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

@@ -102,7 +102,7 @@ public class KafkaConfig {
         // 设置是否自动提交
         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
         // 一次拉取消息数量
-        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);
+        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 40);
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 5*60*1000);
         props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 5*60*1000);
         // 最大处理时间

src/main/java/com/winhc/utils/CommonName.java

@@ -0,0 +1,38 @@
+package com.winhc.utils;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Resources;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2023/5/5 17:28
+ */
+public class CommonName {
+    public static List<String> NAMES = new ArrayList<>();
+
+    static {
+        URL url = Resources.getResource("data/name.txt");
+        try {
+            NAMES = Arrays.stream(Resources.toString(url, Charsets.UTF_8)
+                    .split("\n"))
+                    .filter(StringUtils::isNotBlank)
+                    .distinct().collect(Collectors.toList());
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void main(String[] args) {
+        List<String> n1 = NAMES;
+        System.out.println(n1);
+    }
+}

src/main/java/com/winhc/utils/CompanyUtils.java

@@ -152,6 +152,7 @@ public class CompanyUtils {
     public static List<Map<String, String>> trans(Collection<List<MergePerson>> collect) {
         List<Map<String, String>> list = new ArrayList<>();
         collect.forEach(v -> v.stream()
+                .filter(CompanyUtils::is_pass)
                 .max(Comparator.comparing(MergePerson::getCnt))
                 .ifPresent(maxPerson -> v.forEach(m -> {
                     if (!m.getOld_human_pid().equals(maxPerson.getOld_human_pid())) {
@@ -172,6 +173,14 @@ public class CompanyUtils {
         );
     }
 
+    public static Boolean is_pass(MergePerson m) {
+        //大众人名且超过100家,过滤
+        if (m.getCnt() > 100 && CommonName.NAMES.contains(m.getNew_human_name())) {
+            return false;
+        }
+        return m.getCnt() <= 500;
+    }
+
     public static Collection<List<MergePerson>> getMergePerson(Session session, String cql, HashMap<String, Object> parameters) {
         List<Record> dataList = getRecords(session, cql, parameters);
         return dataList.stream()