fix : IP解析历史数据处理新增线程池调试01

game-module/game-module-manage/src/main/java/com/zanxiang/game/module/manage/ManageApplication.java

@@ -23,7 +23,7 @@ public class ManageApplication {
 
     public static void main(String[] args) {
         SpringApplication.run(ManageApplication.class, args);
-        System.out.println("赞象Manage服务启动成功 < (IP解析历史数据处理新增线程池调试 ･･)ﾉ(._.`) \n" +
+        System.out.println("赞象Manage服务启动成功 < (IP解析历史数据处理新增线程池调试01 ･･)ﾉ(._.`) \n" +
                 "___  ___  ___   _   _   ___  _____  _____ \n" +
                 "|  \\/  | / _ \\ | \\ | | / _ \\|  __ \\|  ___|\n" +
                 "| .  . |/ /_\\ \\|  \\| |/ /_\\ \\ |  \\/| |__  \n" +

game-module/game-module-manage/src/main/java/com/zanxiang/game/module/manage/controller/UserController.java

@@ -166,9 +166,9 @@ public class UserController {
         return ResultVO.ok();
     }
 
-    @ApiOperation(value = "根据主键id解析数据IP")
+    @ApiOperation(value = "根据主键id解析用户数据IP")
     @PatchMapping(value = "/ip/assay/by/ids")
-    @PreAuthorize(permissionKey = "manage:user:ipAssay")
+    @PreAuthorize(permissionKey = "manage:user:ipAssayByIds")
     @ApiResponses(value = {@ApiResponse(code = 200, message = "成功")})
     public ResultVO<Void> userIpAssayById(@RequestParam String ids) {
         ipDataAssayService.userHandleByIds(ids);
@@ -183,4 +183,13 @@ public class UserController {
         ipDataAssayService.userLoginLogHandle();
         return ResultVO.ok();
     }
+
+    @ApiOperation(value = "根据主键id解析玩家登录日志数据IP")
+    @PatchMapping(value = "/login/log/ip/assay/by/ids")
+    @PreAuthorize(permissionKey = "manage:user:loginLogIpAssayByIds")
+    @ApiResponses(value = {@ApiResponse(code = 200, message = "成功")})
+    public ResultVO<Void> userLoginLogHandleByIds(@RequestParam String ids) {
+        ipDataAssayService.userLoginLogHandleByIds(ids);
+        return ResultVO.ok();
+    }
 }

game-module/game-module-manage/src/main/java/com/zanxiang/game/module/manage/service/IIpDataAssayService.java

@@ -10,7 +10,7 @@ import java.util.List;
 public interface IIpDataAssayService {
 
     /**
-     * 根据主键id处理
+     * 根据主键id处理用户数据
      *
      * @param ids : 主键id
      */
@@ -21,6 +21,13 @@ public interface IIpDataAssayService {
      */
     void userHandle(long supperGameId);
 
+    /**
+     * 根据id主键处理登录日志
+     *
+     * @param ids : 主键id
+     */
+    void userLoginLogHandleByIds(String ids);
+
     /**
      * 登录日志处理
      */

game-module/game-module-manage/src/main/java/com/zanxiang/game/module/manage/service/impl/IpDataAssayServiceImpl.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -62,6 +63,16 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
             new ThreadFactoryBuilder()
                     .setNameFormat("sync-ipDataAssay-%d").build());
 
+    private static final ThreadPoolExecutor THREAD_POOL_IP_DATA_ASSAY_QUEUE = new ThreadPoolExecutor(
+            10,
+            10,
+            0,
+            TimeUnit.MINUTES,
+            new LinkedBlockingQueue<>(),
+            new ThreadFactoryBuilder()
+                    .setNameFormat("sync-ipDataAssayQueue-%d").build());
+
+
     @Override
     public void userHandleByIds(String ids) {
         List<Long> userIds = Arrays.stream(ids.split(",")).map(Long::valueOf).collect(Collectors.toList());
@@ -71,7 +82,7 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
         }
         userList.forEach(user -> {
             String ipData = this.ipAssay(user.getIp());
-            log.error("IP解析返回的结果, ipData : {}", ipData);
+            log.error("用户IP解析返回的结果, ipData : {}", ipData);
             if (Strings.isBlank(ipData)) {
                 return;
             }
@@ -112,7 +123,7 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
             log.error("用户单页数据为空, 不处理");
             return;
         }
-        THREAD_POOL_IP_DATA_ASSAY.execute(() -> userList.forEach(user -> {
+        List<CompletableFuture<?>> futures = userList.stream().map(user -> CompletableFuture.runAsync(() -> {
             if (Strings.isNotBlank(user.getIpData())) {
                 return;
             }
@@ -124,7 +135,31 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
             userService.update(new LambdaUpdateWrapper<User>()
                     .set(User::getIpData, ipData)
                     .eq(User::getId, user.getId()));
-        }));
+        }, THREAD_POOL_IP_DATA_ASSAY)).collect(Collectors.toList());
+        try {
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
+        } catch (Exception e) {
+            log.error("等待所有用户处理完成时出现异常: {}", e.getMessage());
+        }
+    }
+
+    @Override
+    public void userLoginLogHandleByIds(String ids) {
+        List<Long> userLoginLogIds = Arrays.stream(ids.split(",")).map(Long::valueOf).collect(Collectors.toList());
+        List<UserLoginLog> userLoginLogList = userLoginLogService.listByIds(userLoginLogIds);
+        if (CollectionUtils.isEmpty(userLoginLogList)) {
+            return;
+        }
+        userLoginLogList.forEach(userLoginLog -> {
+            String ipData = this.ipAssay(userLoginLog.getIp());
+            log.error("登录日志IP解析返回的结果, ipData : {}", ipData);
+            if (Strings.isBlank(ipData)) {
+                return;
+            }
+            userLoginLogService.update(new LambdaUpdateWrapper<UserLoginLog>()
+                    .set(UserLoginLog::getIpData, ipData)
+                    .eq(UserLoginLog::getId, userLoginLog.getId()));
+        });
     }
 
     @Override
@@ -151,7 +186,7 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
             log.error("登录日志单页数据为空, 不处理");
             return;
         }
-        THREAD_POOL_IP_DATA_ASSAY.execute(() -> userLoginLogList.forEach(userLoginLog -> {
+        List<CompletableFuture<?>> futures = userLoginLogList.stream().map(userLoginLog -> CompletableFuture.runAsync(() -> {
             if (Strings.isNotBlank(userLoginLog.getIpData())) {
                 return;
             }
@@ -163,17 +198,23 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
             userLoginLogService.update(new LambdaUpdateWrapper<UserLoginLog>()
                     .set(UserLoginLog::getIpData, ipData)
                     .eq(UserLoginLog::getId, userLoginLog.getId()));
-        }));
+        }, THREAD_POOL_IP_DATA_ASSAY)).collect(Collectors.toList());
+        try {
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
+        } catch (Exception e) {
+            log.error("等待所有登录日志完成时出现异常: {}", e.getMessage());
+        }
     }
 
     @Override
     public void ipDataHandle(List<String> msgList) {
-        msgList.forEach(msg -> {
+        List<CompletableFuture<?>> futures = msgList.stream().map(msg -> CompletableFuture.runAsync(() -> {
+            //数据解析, 消息类型(注册还是登录), 数据主键, ip地址
             String[] array = msg.split(":");
-            //消息类型(注册还是登录), 数据主键, ip地址
             Tuple3<String, String, String> tuple3 = Tuple3.with(array[0], array[1], array[2]);
             String ipData = this.ipAssay(tuple3.third);
             if (Strings.isBlank(ipData)) {
+                log.error("队列消息IP解析结果返回为空, ip : {}", tuple3.third);
                 return;
             }
             //注册用户, t_user表
@@ -190,7 +231,12 @@ public class IpDataAssayServiceImpl implements IIpDataAssayService {
                         .eq(UserLoginLog::getId, Long.parseLong(tuple3.second))
                 );
             }
-        });
+        }, THREAD_POOL_IP_DATA_ASSAY_QUEUE)).collect(Collectors.toList());
+        try {
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
+        } catch (Exception e) {
+            log.error("等待所有队列消息消费完成时出现异常: {}", e.getMessage());
+        }
     }
 
     private String ipAssay(String ip) {

game-module/game-module-manage/src/main/java/com/zanxiang/game/module/manage/task/IpDataAssayTask.java

@@ -47,5 +47,6 @@ public class IpDataAssayTask {
         log.error("IP解析从队列中获取到数据长度 : {}", msgList.size());
         //数据处理执行
         ipDataAssayService.ipDataHandle(msgList);
+        log.error("IP解析队列消息消费完成 : {}", msgList.size());
     }
 }