refactor(douyin): 重构抖音用户信息更新逻辑

- 移除 BaseEntity 中的 createId 字段
- 修改 CreateAndUpdateMetaObjectHandler 中的用户ID处理逻辑，统一使用字符串类型的用户ID
- 将 RemoteDouYinUserInfoService 中的远程调用改为本地Mapper直接访问
- 在抖音工具类中添加用户信息日志记录功能
- 删除旧的服务接口和实现方法，简化代码结构
- 重构抖音直播监控任务，使用线程池并发处理用户数据更新
- 添加详细的线程池配置说明和任务执行统计功能
- 优化数据库查询和更新操作，提高批量处理效率

nexo-api/nexo-api-module/src/main/java/com/nexo/module/api/douyin/RemoteDouYinUserInfoService.java

@@ -11,6 +11,4 @@ import java.util.List;
  */
 public interface RemoteDouYinUserInfoService {
 
-    void updateDouYinUserInfo(NexoDouyinUserInfo nexoDouyinUserInfo);
-    List<NexoDouyinUserInfo> getUpdateInfoList(String secuid);
 }

nexo-api/nexo-api-module/src/main/java/com/nexo/module/api/douyin/utils/DouYinUtils.java

@@ -28,6 +28,7 @@ public class DouYinUtils {
         // 解析json
         try {
             JSONObject reqJson = JSONObject.parseObject(reqStr);
+            log.info("用户信息：{}", reqJson);
             item.setUserId(reqJson.getJSONObject("user").getString("uid"));
             item.setSecUid(reqJson.getJSONObject("user").getString("sec_uid"));
             item.setUniqueId(reqJson.getJSONObject("user").getString("unique_id"));

nexo-common/nexo-common-core/src/main/java/com/nexo/common/core/web/domain/BaseEntity.java

@@ -31,11 +31,6 @@ public class BaseEntity implements Serializable {
     @TableField(exist = false)
     private String searchValue;
 
-    /**
-     * 创建者id
-     */
-    @TableField(fill = FieldFill.INSERT)
-    private Long createId;
     /**
      * 创建者名称
      */

nexo-common/nexo-common-mybatis/src/main/java/com/nexo/common/mybatis/handler/CreateAndUpdateMetaObjectHandler.java

@@ -10,7 +10,6 @@ import com.nexo.common.satoken.utils.LoginHelper;
 import com.nexo.system.api.model.LoginUser;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.ibatis.reflection.MetaObject;
-import org.springframework.util.ObjectUtils;
 
 import java.util.Date;
 
@@ -31,10 +30,7 @@ public class CreateAndUpdateMetaObjectHandler implements MetaObjectHandler {
                 Date current = ObjectUtil.isNotNull(baseEntity.getCreateTime()) ? baseEntity.getCreateTime() : new Date();
                 baseEntity.setCreateTime(current);
                 baseEntity.setUpdateTime(current);
-                String username = StringUtils.isNotBlank(baseEntity.getCreateBy()) ? baseEntity.getCreateBy() : getLoginUsername();
-                Long userId = !ObjectUtils.isEmpty(baseEntity.getCreateId()) ? baseEntity.getCreateId() : getLoginUserId();
-                // 当前已登录 且 创建人ID为空 则填充
-                baseEntity.setCreateId(userId);
+                String username = StringUtils.isNotBlank(baseEntity.getCreateBy()) ? baseEntity.getCreateBy() : getLoginUserId();
                 // 当前已登录 且 创建人为空 则填充
                 baseEntity.setCreateBy(username);
                 // 当前已登录 且 更新人为空 则填充
@@ -53,10 +49,10 @@ public class CreateAndUpdateMetaObjectHandler implements MetaObjectHandler {
                 Date current = new Date();
                 // 更新时间填充(不管为不为空)
                 baseEntity.setUpdateTime(current);
-                String username = getLoginUsername();
+                String userId = getLoginUserId();
                 // 当前已登录 更新人填充(不管为不为空)
-                if (StringUtils.isNotBlank(username)) {
-                    baseEntity.setUpdateBy(username);
+                if (StringUtils.isNotBlank(userId)) {
+                    baseEntity.setUpdateBy(userId);
                 }
             }
         } catch (Exception e) {
@@ -80,14 +76,14 @@ public class CreateAndUpdateMetaObjectHandler implements MetaObjectHandler {
     /**
      * 获取登录用户名
      */
-    private Long getLoginUserId() {
+    private String getLoginUserId() {
         LoginUser loginUser;
         try {
             loginUser = LoginHelper.getLoginUser();
         } catch (Exception e) {
             return null;
         }
-        return ObjectUtil.isNotNull(loginUser) ? loginUser.getUserId() : null;
+        return ObjectUtil.isNotNull(loginUser) ? String.valueOf(loginUser.getUserId()) : null;
     }
 
 }

nexo-example/nexo-model/src/main/java/com/nexo/model/douyin/dubbo/RemoteDouYinUserInfoServiceImpl.java

@@ -21,15 +21,4 @@ import java.util.List;
 @DubboService
 public class RemoteDouYinUserInfoServiceImpl implements RemoteDouYinUserInfoService {
 
-    private final INexoDouyinUserInfoService userInfoService;
-
-    @Override
-    public void updateDouYinUserInfo(NexoDouyinUserInfo nexoDouyinUserInfo) {
-        userInfoService.updateBy(nexoDouyinUserInfo);
-    }
-
-    @Override
-    public List<NexoDouyinUserInfo> getUpdateInfoList(String secuid) {
-        return userInfoService.getUpdateInfoList(secuid);
-    }
 }

nexo-example/nexo-model/src/main/java/com/nexo/model/douyin/mapper/NexoDouyinUserInfoMapper.java

@@ -1,23 +1,17 @@
 package com.nexo.model.douyin.mapper;
 
-import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
 import com.nexo.common.mybatis.annotation.DataColumn;
 import com.nexo.common.mybatis.annotation.DataPermission;
 import com.nexo.common.mybatis.core.mapper.BaseMapperPlus;
 import com.nexo.module.api.douyin.domain.NexoDouyinUserInfo;
 
-import java.util.List;
-
 /**
  * 抖音用户信息Mapper接口
  *
  * @author nexo
  * @date 2026-05-21
  */
-@DataPermission({@DataColumn(key = "deptName", value = "dept_id"), @DataColumn(key = "userName", value = "create_id")})
+@DataPermission({@DataColumn(key = "deptName", value = "dept_id"), @DataColumn(key = "userName", value = "create_by")})
 public interface NexoDouyinUserInfoMapper extends BaseMapperPlus<NexoDouyinUserInfoMapper, NexoDouyinUserInfo, NexoDouyinUserInfo> {
 
-    @InterceptorIgnore(dataPermission = "true")  // 忽略数据权限
-    List<NexoDouyinUserInfo> getUpdateInfoList(String secuid);
-
 }

nexo-example/nexo-model/src/main/java/com/nexo/model/douyin/service/INexoDouyinUserInfoService.java

@@ -48,5 +48,4 @@ public interface INexoDouyinUserInfoService {
 
     Boolean updateMonitorState(UpdataStatusBo item);
 
-    List<NexoDouyinUserInfo> getUpdateInfoList(String secuid);
 }

nexo-example/nexo-model/src/main/java/com/nexo/model/douyin/service/impl/NexoDouyinUserInfoServiceImpl.java

@@ -153,8 +153,4 @@ public class NexoDouyinUserInfoServiceImpl implements INexoDouyinUserInfoService
         return baseMapper.updateById(userInfo) > 0;
     }
 
-    @Override
-    public List<NexoDouyinUserInfo> getUpdateInfoList(String secuid) {
-        return baseMapper.getUpdateInfoList(secuid);
-    }
 }

nexo-example/nexo-model/src/main/resources/mapper/douyin/NexoDouyinUserInfoMapper.xml

@@ -22,16 +22,4 @@
         <result property="roomId" column="room_id"/>
     </resultMap>
 
-    <select id="getUpdateInfoList" resultType="com.nexo.module.api.douyin.domain.NexoDouyinUserInfo">
-        SELECT * FROM `nexo_douyin_user_info` WHERE job_info_status = 1
-        <where>
-            <if test="secuid != null">
-                AND sec_uid = #{secuid}
-            </if>
-            <if test="secuid == null">
-                group by create_by
-            </if>
-        </where>
-    </select>
-
 </mapper>

nexo-modules/nexo-job/src/main/java/com/nexo/job/mapper/NexoDouyinUserInfoJobMapper.java

@@ -0,0 +1,17 @@
+package com.nexo.job.mapper;
+
+import com.nexo.common.mybatis.core.mapper.BaseMapperPlus;
+import com.nexo.module.api.douyin.domain.NexoDouyinUserInfo;
+
+import java.util.List;
+
+/**
+ * 抖音用户信息Mapper接口
+ *
+ * @author nexo
+ * @date 2026-05-21
+ */
+public interface NexoDouyinUserInfoJobMapper extends BaseMapperPlus<NexoDouyinUserInfoJobMapper, NexoDouyinUserInfo, NexoDouyinUserInfo> {
+
+    List<NexoDouyinUserInfo> getUpdateInfoList(String secuid);
+}

nexo-modules/nexo-job/src/main/java/com/nexo/job/service/SampleService.java

@@ -1,16 +1,18 @@
 package com.nexo.job.service;
 
-import com.alibaba.fastjson2.JSONObject;
-import com.nexo.module.api.douyin.RemoteDouYinUserInfoService;
+import com.nexo.job.mapper.NexoDouyinUserInfoJobMapper;
 import com.nexo.module.api.douyin.domain.NexoDouyinUserInfo;
 import com.nexo.module.api.douyin.utils.DouYinUtils;
-import com.xxl.job.core.context.XxlJobContext;
+import com.xxl.job.core.context.XxlJobHelper;
 import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.dubbo.config.annotation.DubboReference;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.Resource;
+import java.util.Date;
 import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 /**
@@ -28,26 +30,128 @@ import java.util.stream.Collectors;
 @Service
 public class SampleService {
 
-    @DubboReference
-    private RemoteDouYinUserInfoService remoteDouYinUserInfoService;
+    @Resource
+    private NexoDouyinUserInfoJobMapper nexoDouyinUserInfoMapper;
+    /**
+     * 抖音直播监控任务专用线程池
+     * <p>
+     * 线程池配置说明：
+     * - 核心线程数：5，保持常驻线程数量
+     * - 最大线程数：10，高峰期可扩展的最大线程数
+     * - 空闲线程存活时间：60秒，超过核心线程数的空闲线程将在60秒后回收
+     * - 工作队列：有界阻塞队列，容量1000，用于缓存待处理任务
+     * - 线程工厂：自定义线程命名，便于日志追踪和问题定位
+     * - 拒绝策略：CallerRunsPolicy，当队列满且线程达最大值时，由调用线程执行任务，避免任务丢失
+     * <p>
+     * 适用场景：
+     * - 批量处理抖音用户数据更新
+     * - 并发执行网络请求和数据库操作
+     * - 提高定时任务执行效率
+     */
+    private static final ExecutorService executor = new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), new ThreadFactory() {
+        private final AtomicInteger threadNumber = new AtomicInteger(1);
+
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread t = new Thread(r, "douyin-live-monitor-" + threadNumber.getAndIncrement());
+            t.setDaemon(false);
+            return t;
+        }
+    }, new ThreadPoolExecutor.CallerRunsPolicy());
 
     /**
-     * 抖音直播监控
+     * 抖音直播监控定时任务
+     * <p>
+     * 功能说明：
+     * 1. 获取所有需要更新的抖音用户信息列表
+     * 2. 使用线程池并发处理每个用户的数据更新
+     * 3. 调用抖音API获取最新用户信息
+     * 4. 批量更新数据库中对应的用户记录
+     * 5. 统计成功和失败数量，记录执行日志
+     * <p>
+     * 并发策略：
+     * - 使用预配置的线程池（核心30线程，最大50线程）并发执行
+     * - 通过CountDownLatch等待所有任务完成
+     * - 使用AtomicInteger保证计数的线程安全
+     * <p>
+     * 异常处理：
+     * - 单个用户处理失败不影响其他用户
+     * - 详细记录失败原因便于问题排查
+     * - 捕获中断异常并恢复中断状态
+     *
+     * @throws Exception 任务执行异常
      */
     @XxlJob("douyinLiveMonitor")
     public void douyinLiveMonitor() throws Exception {
-        List<NexoDouyinUserInfo> updateInfoList = remoteDouYinUserInfoService.getUpdateInfoList(null);
+        long startTime = new Date().getTime();
+        List<NexoDouyinUserInfo> updateInfoList = nexoDouyinUserInfoMapper.getUpdateInfoList(null);
+        XxlJobHelper.log("更新用户数：{}", updateInfoList.size());
+
+        if (updateInfoList.isEmpty()) {
+            XxlJobHelper.log("没有需要更新的用户数据");
+            return;
+        }
+
+        AtomicInteger successCount = new AtomicInteger(0);
+        AtomicInteger failCount = new AtomicInteger(0);
+        CountDownLatch latch = new CountDownLatch(updateInfoList.size());
+
         for (int i = 0; i < updateInfoList.size(); i++) {
-            NexoDouyinUserInfo info = updateInfoList.get(i);
-            List<String> strings = remoteDouYinUserInfoService.getUpdateInfoList(info.getSecUid()).stream().map(item -> item.getSecUid()).collect(Collectors.toList());
+            final NexoDouyinUserInfo info = updateInfoList.get(i);
+            final int index = i + 1;
 
+            executor.submit(() -> {
+                try {
+                    long time1 = new Date().getTime();
+                    XxlJobHelper.log("开始处理第{}条数据：{} - {} - {}", index, info.getUserId(), info.getNickname(), info.getSecUid());
+
+                    NexoDouyinUserInfo userInfo = DouYinUtils.getUserInfo(info.getSecUid());
+                    List<Long> longs = nexoDouyinUserInfoMapper.getUpdateInfoList(info.getSecUid()).stream().map(item -> item.getId()).collect(Collectors.toList());
+
+                    XxlJobHelper.log("user_id {} 昵称 {} 需要更新 {} 条用户记录", info.getUserId(), info.getNickname(), longs.size());
+
+                    for (Long id : longs) {
+                        try {
+                            userInfo.setId(id);
+                            userInfo.setJobInfoStatus(null);
+                            userInfo.setJobLiveStatus(null);
+                            userInfo.setJobVideoStatus(null);
+                            userInfo.setCreateBy(null);
+                            userInfo.setCreateTime(null);
+                            userInfo.setDeptId(null);
+                            userInfo.setUpdateTime(new Date());
+
+                            nexoDouyinUserInfoMapper.updateById(userInfo);
+                            successCount.incrementAndGet();
+                        } catch (Exception e) {
+                            failCount.incrementAndGet();
+                            XxlJobHelper.log("更新用户记录失败，ID: {}, 错误: {}", id, e.getMessage());
+                            log.error("更新用户记录失败，ID: {}", id, e);
+                        }
+                    }
+
+                    long costTime = (new Date().getTime() - time1) / 1000;
+                    XxlJobHelper.log("第{}条数据处理完成，耗时 {} 秒", index, costTime);
+                } catch (Exception e) {
+                    failCount.incrementAndGet();
+                    XxlJobHelper.log("处理第{}条数据失败：{}, 错误: {}", index, info.getSecUid(), e.getMessage());
+                    log.error("处理第{}条数据失败：{}", index, info.getSecUid(), e);
+                } finally {
+                    latch.countDown();
+                }
+            });
+        }
+
+        try {
+            latch.await();
+            long totalTime = (new Date().getTime() - startTime) / 1000;
+            XxlJobHelper.log("所有数据处理完成！总计: {}, 成功: {}, 失败: {}, 总耗时: {} 秒", updateInfoList.size(), successCount.get(), failCount.get(), totalTime);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            XxlJobHelper.log("任务被中断：{}", e.getMessage());
+            log.error("任务被中断", e);
         }
-        // 通过上下文获取任务参数
-        String param = XxlJobContext.getXxlJobContext().getJobParam();
-        JSONObject parsed = JSONObject.parseObject(param);
-        String sec_uid = parsed.getString("sec_uid");
-        NexoDouyinUserInfo userInfo = DouYinUtils.getUserInfo(sec_uid);
-        remoteDouYinUserInfoService.updateDouYinUserInfo(userInfo);
     }
 
+
 }

nexo-modules/nexo-job/src/main/resources/mapper/douyin/NexoDouyinUserInfoJobMapper.xml

@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper
+        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
+        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.nexo.job.mapper.NexoDouyinUserInfoJobMapper">
+
+    <resultMap type="com.nexo.module.api.douyin.domain.NexoDouyinUserInfo" id="NexoDouyinUserInfoResult">
+        <result property="userId" column="user_id"/>
+        <result property="secUid" column="sec_uid"/>
+        <result property="uniqueId" column="unique_id"/>
+        <result property="nickname" column="nickname"/>
+        <result property="signature" column="signature"/>
+        <result property="ipLocation" column="ip_location"/>
+        <result property="avatar" column="avatar"/>
+        <result property="liveStatus" column="live_status"/>
+        <result property="userAge" column="user_age"/>
+        <result property="totalFavorited" column="total_favorited"/>
+        <result property="followerCount" column="follower_count"/>
+        <result property="followingCount" column="following_count"/>
+        <result property="awemeCount" column="aweme_count"/>
+        <result property="coverUrl" column="cover_url"/>
+        <result property="roomId" column="room_id"/>
+    </resultMap>
+
+    <select id="getUpdateInfoList" resultType="com.nexo.module.api.douyin.domain.NexoDouyinUserInfo">
+        SELECT * FROM `nexo_douyin_user_info` WHERE job_info_status = 1
+        <if test="secuid != null">
+            AND sec_uid = #{secuid}
+        </if>
+        <if test="secuid == null">
+            group by user_id
+        </if>
+    </select>
+
+</mapper>