获取手机号
This commit is contained in:
@@ -0,0 +1,310 @@
|
|||||||
|
package com.budwk.app.sys.services;
|
||||||
|
|
||||||
|
import cn.hutool.core.util.StrUtil;
|
||||||
|
import cn.hutool.json.JSONUtil;
|
||||||
|
import com.budwk.app.base.constant.RedisConstant;
|
||||||
|
import com.budwk.app.base.constant.RoleConstant;
|
||||||
|
import com.budwk.app.web.commons.auth.utils.AuthUtil;
|
||||||
|
import com.budwk.app.base.exception.BaseException;
|
||||||
|
import com.budwk.app.sys.models.Sys_user;
|
||||||
|
import com.budwk.app.sys.param.SysDataUserUpdatePageForm;
|
||||||
|
import org.nutz.aop.interceptor.ioc.TransAop;
|
||||||
|
import org.nutz.dao.Cnd;
|
||||||
|
import org.nutz.dao.Dao;
|
||||||
|
import org.nutz.dao.Sqls;
|
||||||
|
import org.nutz.dao.sql.Sql;
|
||||||
|
import org.nutz.integration.jedis.RedisService;
|
||||||
|
import org.nutz.ioc.aop.Aop;
|
||||||
|
import org.nutz.ioc.loader.annotation.Inject;
|
||||||
|
import org.nutz.ioc.loader.annotation.IocBean;
|
||||||
|
import org.nutz.json.Json;
|
||||||
|
import org.nutz.lang.util.NutMap;
|
||||||
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||||
|
|
||||||
|
import java.util.*;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 手机号补齐任务：工号去重、会员优先、单并发限速，Redis保存队列和检查点。
|
||||||
|
* 编排线程不持有数据库事务；每人工号通过注入的拉取service以独立短事务写入。
|
||||||
|
*/
|
||||||
|
@IocBean
|
||||||
|
public class SysUserMobileSyncService {
|
||||||
|
// 同一hash tag使任务原子脚本兼容Redis Cluster；记录保留七天，不保存手机号。
|
||||||
|
private static final String PREFIX = RedisConstant.PLATFORM_REDIS_PREFIX + "{userMobileSync}:";
|
||||||
|
private static final String LOCK = PREFIX + "lock";
|
||||||
|
private static final String LATEST = PREFIX + "latest";
|
||||||
|
private static final String EMPTY = PREFIX + "empty";
|
||||||
|
private static final int RETAIN_SECONDS = 7 * 24 * 3600;
|
||||||
|
private static final int LEASE_SECONDS = 120;
|
||||||
|
@Inject private Dao dao;
|
||||||
|
@Inject private RedisService redisService;
|
||||||
|
@Inject private SysDataUserPullService sysDataUserPullService;
|
||||||
|
@Inject private ThreadPoolTaskExecutor threadPoolTaskExecutor;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 预览待请求工号范围，不调用外部接口。
|
||||||
|
* @param scope ALL全部、MEMBER仅会员、FILTER当前筛选的全部结果
|
||||||
|
* @param mode FILL仅补空号、REFRESH核对范围内所有人员
|
||||||
|
* @param form 姓名、工号、单位、在职状态、编制类别、教职工类别；不使用分页参数
|
||||||
|
* @return scopeCount范围工号数、existingCount已有号码数、cooldownCount冷却数、total待请求数
|
||||||
|
*/
|
||||||
|
public NutMap preview(String scope, String mode, SysDataUserUpdatePageForm form) {
|
||||||
|
checkAccess();
|
||||||
|
return select(scope, mode, form, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 启动后台任务，返回脱敏进度；同一时间只允许一个本功能任务。
|
||||||
|
* @param scope ALL/MEMBER/FILTER
|
||||||
|
* @param mode FILL/REFRESH
|
||||||
|
* @param form 当前列表筛选条件（仅FILTER使用）
|
||||||
|
* @param sourceTaskId 继续或重试的原任务编号；新任务传空
|
||||||
|
* @param action NEW新任务、RESUME继续未处理部分、RETRY仅失败部分
|
||||||
|
* @return taskId、status及total/processed/updated/empty/failed/skipped统计
|
||||||
|
*/
|
||||||
|
@Aop(TransAop.READ_COMMITTED)
|
||||||
|
public NutMap start(String scope, String mode, SysDataUserUpdatePageForm form, String sourceTaskId, String action) {
|
||||||
|
checkAccess();
|
||||||
|
List<String> queue;
|
||||||
|
if ("NEW".equals(action)) {
|
||||||
|
NutMap selection = select(scope, mode, form, true);
|
||||||
|
queue = (List<String>) selection.get("queue");
|
||||||
|
} else if ("RESUME".equals(action) || "RETRY".equals(action)) {
|
||||||
|
NutMap source = read(sourceTaskId);
|
||||||
|
if (source.isEmpty()) throw new BaseException("任务记录已过期，请重新预览");
|
||||||
|
if (sourceTaskId.equals(redisService.get(LOCK))) throw new BaseException("原任务仍在执行");
|
||||||
|
mode = source.getString("mode");
|
||||||
|
scope = source.getString("scope");
|
||||||
|
if ("RETRY".equals(action)) {
|
||||||
|
queue = strings(source.get("failedJobs"));
|
||||||
|
} else {
|
||||||
|
List<String> original = queue(sourceTaskId);
|
||||||
|
queue = new ArrayList<>(original.subList(Math.min(source.getInt("processed", 0), original.size()), original.size()));
|
||||||
|
// 继续时将原任务失败项排在未处理项之后，避免切换最近任务后遗失失败重试入口。
|
||||||
|
queue.addAll(strings(source.get("failedJobs")));
|
||||||
|
queue = new ArrayList<>(new LinkedHashSet<>(queue));
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
throw new BaseException("不支持的任务操作");
|
||||||
|
}
|
||||||
|
if (queue.isEmpty()) throw new BaseException("没有需要处理的人员");
|
||||||
|
String id = UUID.randomUUID().toString();
|
||||||
|
NutMap state = NutMap.NEW().addv("taskId", id).addv("status", "RUNNING")
|
||||||
|
.addv("scope", scope).addv("mode", mode).addv("total", queue.size())
|
||||||
|
.addv("processed", 0).addv("updated", 0).addv("empty", 0).addv("failed", 0)
|
||||||
|
.addv("skipped", 0).addv("failedJobs", new ArrayList<String>())
|
||||||
|
.addv("message", "后台拉取中，每秒最多一个工号请求")
|
||||||
|
.addv("startedAt", System.currentTimeMillis());
|
||||||
|
// 锁、队列、初始进度同时写入，防止重复点击及进程中途退出留下半个任务。
|
||||||
|
String script = "if redis.call('exists',KEYS[1]) == 1 then return 0 end "
|
||||||
|
+ "redis.call('setex',KEYS[1],ARGV[2],ARGV[1]); "
|
||||||
|
+ "redis.call('setex',KEYS[2],ARGV[3],ARGV[1]); "
|
||||||
|
+ "redis.call('setex',KEYS[3],ARGV[3],ARGV[4]); "
|
||||||
|
+ "redis.call('setex',KEYS[4],ARGV[3],ARGV[5]); return 1";
|
||||||
|
Object acquired = redisService.eval(script, Arrays.asList(LOCK, LATEST, stateKey(id), PREFIX + "queue:" + id),
|
||||||
|
Arrays.asList(id, String.valueOf(LEASE_SECONDS), String.valueOf(RETAIN_SECONDS), Json.toJson(state), Json.toJson(queue)));
|
||||||
|
if (!Long.valueOf(1).equals(acquired)) throw new BaseException("已有手机号任务执行中，请查看当前进度");
|
||||||
|
NutMap initial = publicState(state);
|
||||||
|
final List<String> jobs = queue;
|
||||||
|
try {
|
||||||
|
threadPoolTaskExecutor.execute(() -> run(state, jobs));
|
||||||
|
} catch (RuntimeException e) {
|
||||||
|
state.put("status", "INTERRUPTED");
|
||||||
|
state.put("message", "后台执行资源繁忙，请稍后继续任务");
|
||||||
|
checkpoint(state, true);
|
||||||
|
throw new BaseException("后台执行资源繁忙，请稍后继续任务");
|
||||||
|
}
|
||||||
|
return initial;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 查询指定任务；taskId为空读取最近任务。锁过期的运行任务显示中断，可手动继续。
|
||||||
|
* @param taskId 后端返回的UUID任务编号或空值
|
||||||
|
* @return 进度对象，不含工号队列、手机号或失败人员明细
|
||||||
|
*/
|
||||||
|
public NutMap status(String taskId) {
|
||||||
|
checkAccess();
|
||||||
|
String id = StrUtil.isBlank(taskId) ? redisService.get(LATEST) : taskId;
|
||||||
|
NutMap state = read(id);
|
||||||
|
if (state.isEmpty()) return state;
|
||||||
|
if ("RUNNING".equals(state.getString("status")) && !id.equals(redisService.get(LOCK))) {
|
||||||
|
state.put("status", "INTERRUPTED");
|
||||||
|
state.put("message", "任务心跳已中断，可继续未完成部分；已保存的号码不会丢失");
|
||||||
|
} else if ("RUNNING".equals(state.getString("status")) && "1".equals(redisService.get(PREFIX + "stop:" + id))) {
|
||||||
|
state.put("message", "正在停止，等待当前工号处理结束");
|
||||||
|
}
|
||||||
|
return publicState(state);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 请求停止指定任务；当前工号处理结束后保存检查点，已完成结果不回滚。
|
||||||
|
* @param taskId 当前任务UUID
|
||||||
|
* @return 最新脱敏进度对象
|
||||||
|
*/
|
||||||
|
@Aop(TransAop.READ_COMMITTED)
|
||||||
|
public NutMap stop(String taskId) {
|
||||||
|
checkAccess();
|
||||||
|
validateId(taskId);
|
||||||
|
if (taskId.equals(redisService.get(LOCK))) redisService.setex(PREFIX + "stop:" + taskId, RETAIN_SECONDS, "1");
|
||||||
|
return status(taskId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 按工号去重并优先会员；读取冷却时间，不在预览中发起远程调用。 */
|
||||||
|
private NutMap select(String scope, String mode, SysDataUserUpdatePageForm form, boolean includeQueue) {
|
||||||
|
if (!Arrays.asList("ALL", "MEMBER", "FILTER").contains(scope)
|
||||||
|
|| !Arrays.asList("FILL", "REFRESH").contains(mode)) throw new BaseException("手机号拉取范围或模式无效");
|
||||||
|
Cnd cnd = Cnd.where("u.loginname", "IS NOT", null).and("TRIM(u.loginname)", "<>", "");
|
||||||
|
if ("MEMBER".equals(scope)) cnd.and("u.member", "=", true);
|
||||||
|
if ("FILTER".equals(scope)) {
|
||||||
|
if (form == null) throw new BaseException("缺少列表筛选条件");
|
||||||
|
cnd.and(Cnd.likeEX("u.username", form.getUserName()));
|
||||||
|
cnd.and(Cnd.likeEX("u.loginname", form.getLoginName()));
|
||||||
|
cnd.andEX("u.unitId", "=", form.getUnitId());
|
||||||
|
cnd.andEX("u.userState", "=", form.getUserState());
|
||||||
|
cnd.andEX("u.preparedBy", "=", form.getPreparedBy());
|
||||||
|
cnd.andEX("u.personType", "=", form.getPersonType());
|
||||||
|
}
|
||||||
|
Sql sql = Sqls.create("SELECT u.loginname, MAX(COALESCE(u.member,0)) AS memberFirst, "
|
||||||
|
+ "MIN(CASE WHEN u.mobile IS NULL OR TRIM(u.mobile) = '' THEN 0 ELSE 1 END) AS hasMobile "
|
||||||
|
+ "FROM vw_user u $condition GROUP BY u.loginname ORDER BY memberFirst DESC,u.loginname");
|
||||||
|
sql.setCondition(cnd);
|
||||||
|
sql.setCallback(Sqls.callback.maps());
|
||||||
|
dao.execute(sql);
|
||||||
|
Map<String, String> emptyTimes = redisService.hgetAll(EMPTY);
|
||||||
|
int existing = 0, cooldown = 0;
|
||||||
|
List<String> jobs = new ArrayList<>();
|
||||||
|
List<NutMap> rows = sql.getList(NutMap.class);
|
||||||
|
for (NutMap row : rows) {
|
||||||
|
String jobNo = row.getString("loginname");
|
||||||
|
boolean hasMobile = row.getInt("hasMobile", 0) == 1;
|
||||||
|
if (hasMobile) existing++;
|
||||||
|
if ("FILL".equals(mode) && hasMobile) continue;
|
||||||
|
if (inCooldown(emptyTimes.get(jobNo))) { cooldown++; continue; }
|
||||||
|
jobs.add(jobNo);
|
||||||
|
}
|
||||||
|
NutMap result = NutMap.NEW().addv("scopeCount", rows.size()).addv("existingCount", existing)
|
||||||
|
.addv("cooldownCount", cooldown).addv("total", jobs.size());
|
||||||
|
if (includeQueue) result.put("queue", jobs);
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 编排只负责远程读取与检查点；每人独立提交，连续五次失败暂停，支持手动继续和失败重试。 */
|
||||||
|
private void run(NutMap state, List<String> jobs) {
|
||||||
|
String id = state.getString("taskId");
|
||||||
|
boolean fillOnly = "FILL".equals(state.getString("mode"));
|
||||||
|
int consecutiveFailures = 0;
|
||||||
|
try {
|
||||||
|
for (String jobNo : jobs) {
|
||||||
|
if (!checkpoint(state, false)) return;
|
||||||
|
if ("1".equals(redisService.get(PREFIX + "stop:" + id))) {
|
||||||
|
state.put("status", "STOPPED");
|
||||||
|
state.put("message", "已停止，可继续未完成部分");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if (inCooldown(redisService.hget(EMPTY, jobNo))) {
|
||||||
|
increment(state, "skipped");
|
||||||
|
} else {
|
||||||
|
List<Sys_user> users = dao.query(Sys_user.class, Cnd.where(Sys_user::getLoginname, "=", jobNo));
|
||||||
|
Map<String, String> originals = new LinkedHashMap<>();
|
||||||
|
for (Sys_user user : users) {
|
||||||
|
if (!fillOnly || StrUtil.isBlank(user.getMobile())) originals.put(user.getId(), user.getMobile());
|
||||||
|
}
|
||||||
|
if (originals.isEmpty()) {
|
||||||
|
increment(state, "skipped");
|
||||||
|
} else {
|
||||||
|
try {
|
||||||
|
String mobile = sysDataUserPullService.fetchTeacherMobile(jobNo);
|
||||||
|
// 外部请求返回后再次核对租约，过期的旧worker不得继续落库或提交检查点。
|
||||||
|
if (!checkpoint(state, false)) return;
|
||||||
|
if (StrUtil.isBlank(mobile)) {
|
||||||
|
redisService.hset(EMPTY, jobNo, String.valueOf(System.currentTimeMillis()));
|
||||||
|
redisService.expire(EMPTY, RETAIN_SECONDS);
|
||||||
|
increment(state, "empty");
|
||||||
|
} else {
|
||||||
|
int changed = sysDataUserPullService.saveTeacherMobile(jobNo, mobile, originals, fillOnly);
|
||||||
|
increment(state, changed > 0 ? "updated" : "skipped");
|
||||||
|
}
|
||||||
|
consecutiveFailures = 0;
|
||||||
|
} catch (Exception e) {
|
||||||
|
// 不将第三方异常原文返回前端或写日志，避免泄露响应中的手机号和凭据。
|
||||||
|
increment(state, "failed");
|
||||||
|
((List<String>) state.get("failedJobs")).add(jobNo);
|
||||||
|
consecutiveFailures++;
|
||||||
|
}
|
||||||
|
// 每人工号请求结束后至少等待一秒；失败逐步退避，五连败后暂停。
|
||||||
|
if (consecutiveFailures < 5) Thread.sleep(Math.max(1000, consecutiveFailures * 2000L));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
increment(state, "processed");
|
||||||
|
if (consecutiveFailures >= 5) {
|
||||||
|
state.put("status", "PAUSED");
|
||||||
|
state.put("message", "连续五次拉取失败，已暂停；请检查接口后继续或重试失败项");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if (!checkpoint(state, false)) return;
|
||||||
|
}
|
||||||
|
if ("RUNNING".equals(state.getString("status"))) {
|
||||||
|
state.put("status", "COMPLETED");
|
||||||
|
state.put("message", state.getInt("failed", 0) > 0 ? "拉取结束，部分失败可单独重试" : "拉取完成");
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
if (e instanceof InterruptedException) Thread.currentThread().interrupt();
|
||||||
|
state.put("status", "INTERRUPTED");
|
||||||
|
state.put("message", "任务执行中断，可继续未完成部分");
|
||||||
|
} finally {
|
||||||
|
checkpoint(state, true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 原子校验锁持有者后保存进度及续期；旧任务不能覆盖新任务，也不能释放别人的锁。 */
|
||||||
|
private boolean checkpoint(NutMap state, boolean finish) {
|
||||||
|
String id = state.getString("taskId");
|
||||||
|
state.put("updatedAt", System.currentTimeMillis());
|
||||||
|
String script = "if redis.call('get',KEYS[1]) ~= ARGV[1] then return 0 end "
|
||||||
|
+ "redis.call('setex',KEYS[2],ARGV[2],ARGV[3]); "
|
||||||
|
+ "redis.call('expire',KEYS[3],ARGV[2]); redis.call('expire',KEYS[4],ARGV[2]); "
|
||||||
|
+ (finish ? "redis.call('del',KEYS[1]); " : "redis.call('expire',KEYS[1],ARGV[4]); ") + "return 1";
|
||||||
|
return Long.valueOf(1).equals(redisService.eval(script,
|
||||||
|
Arrays.asList(LOCK, stateKey(id), PREFIX + "queue:" + id, LATEST),
|
||||||
|
Arrays.asList(id, String.valueOf(RETAIN_SECONDS), Json.toJson(state), String.valueOf(LEASE_SECONDS))));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 全局人员拉取仅开放给系统管理员和校工会管理员，沿用现有会员手机号入口的角色限制。 */
|
||||||
|
private void checkAccess() {
|
||||||
|
if (!AuthUtil.hasRoleOr(RoleConstant.SYSADMIN.name(), RoleConstant.SCHOOL_UNION_ADMIN.name())) {
|
||||||
|
throw new BaseException("当前用户无权拉取系统人员手机号");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private boolean inCooldown(String time) {
|
||||||
|
if (time == null) return false;
|
||||||
|
try { return System.currentTimeMillis() - Long.parseLong(time) < RETAIN_SECONDS * 1000L; }
|
||||||
|
catch (NumberFormatException e) { return false; }
|
||||||
|
}
|
||||||
|
|
||||||
|
private void increment(NutMap state, String field) { state.put(field, state.getInt(field, 0) + 1); }
|
||||||
|
private String stateKey(String id) { return PREFIX + "state:" + id; }
|
||||||
|
private void validateId(String id) {
|
||||||
|
if (id == null || !id.matches("[0-9a-fA-F-]{36}")) throw new BaseException("任务编号无效");
|
||||||
|
}
|
||||||
|
private NutMap read(String id) {
|
||||||
|
if (StrUtil.isBlank(id)) return NutMap.NEW();
|
||||||
|
validateId(id);
|
||||||
|
String json = redisService.get(stateKey(id));
|
||||||
|
return json == null ? NutMap.NEW() : Json.fromJson(NutMap.class, json);
|
||||||
|
}
|
||||||
|
private List<String> queue(String id) {
|
||||||
|
String json = redisService.get(PREFIX + "queue:" + id);
|
||||||
|
if (json == null) throw new BaseException("任务队列已过期，请重新预览");
|
||||||
|
return JSONUtil.toList(JSONUtil.parseArray(json), String.class);
|
||||||
|
}
|
||||||
|
private List<String> strings(Object value) {
|
||||||
|
return value == null ? new ArrayList<>() : JSONUtil.toList(JSONUtil.parseArray(value), String.class);
|
||||||
|
}
|
||||||
|
private NutMap publicState(NutMap state) {
|
||||||
|
NutMap result = NutMap.NEW();
|
||||||
|
result.putAll(state);
|
||||||
|
result.remove("failedJobs");
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user