package com.example.demo.Mysql; import com.example.demo.Util.BaseDES; import com.example.demo.domain.entity.User; import com.example.demo.service.coin.AdminService; import com.example.demo.service.coin.MarketService; import com.example.demo.service.coin.UserService; import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.http.*; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.client.RestTemplate; import javax.sql.DataSource; import java.math.BigDecimal; import java.sql.*; import java.time.LocalDateTime; import java.time.Month; import java.time.format.DateTimeFormatter; import java.util.*; @Service public class MysqlServiceImpl implements MysqlService { @Autowired private RestTemplate restTemplate; Set validZeroTypes = new HashSet<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 11, 12, 13, 14, 18, 19, 20, 21, 22, 23, 24, 26, 28, 29, 35, 36, 40, 45, 46, 47, 48, 49, 53, 54, 60)); Set validOneTypes = new HashSet<>(Arrays.asList(9, 15, 17, 25, 27, 37, 41, 42, 43, 50, 51, 62)); Set validTwoTypes = new HashSet<>(Arrays.asList(52,61)); Set validThreeTypes = new HashSet<>(Arrays.asList(10, 16, 30, 31, 32, 33, 34, 39, 44)); Set validFourTypes = new HashSet<>(Arrays.asList(55, 56, 57, 58, 59, 63, 64, 65)); LocalDateTime now = LocalDateTime.now(); Month currentMonth = now.getMonth(); @Autowired private AdminService adminService; @Autowired private UserService userService; @Autowired private MarketService marketService; @Autowired @Qualifier("sqlserver1DataSource") private DataSource sqlserver1DataSource; @Autowired @Qualifier("mysql1DataSource") private DataSource mysql1DataSource; private static final Logger logger = LoggerFactory.getLogger(MysqlServiceImpl.class); @Override @Transactional(transactionManager = "mysqlTransactionManager") // 👈 保证插入和用户更新在一个事务 public void getSqlserverData() throws Exception { logger.info("开始从 SQL Server 同步数据到 MySQL"); try (Connection sqlServerConn = sqlserver1DataSource.getConnection(); Connection mysqlConn = mysql1DataSource.getConnection()) { logger.info("开始查询数据..."); int pageSize = 100; int offset = 0; boolean hasMoreData = true; // 👇 恢复动态时间查询(原注释掉的硬编码时间已移除) String querySql = """ SELECT id, gtype, jwcode, free, core_jb, buy_jb, cz_time, cz_user, cz_bz, operation_platform, goods_name FROM hwhcGold.dbo.user_gold_records WHERE cz_time >= ? ORDER BY cz_time ASC OFFSET ? ROWS FETCH NEXT ? ROWS ONLY; """; String insertSql = """ INSERT IGNORE INTO user_gold_record (order_code, jwcode, sum_gold, permanent_gold, free_june, free_december, task_gold, pay_platform, goods_name, refund_type, refund_model, remark, type, admin_id, audit_status, create_time, flag, update_time, audit_time, is_refund, uid) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """; while (hasMoreData) { try (PreparedStatement sqlServerStmt = sqlServerConn.prepareStatement(querySql)) { // 👇 恢复动态时间参数 sqlServerStmt.setTimestamp(1, Timestamp.valueOf(LocalDateTime.now().minusHours(1))); sqlServerStmt.setInt(2, offset); sqlServerStmt.setInt(3, pageSize); ResultSet resultSet = sqlServerStmt.executeQuery(); if (!resultSet.next()) { logger.info("无更多数据,offset={}", offset); break; } // 👇 步骤1:收集本批次所有记录 List batchRecords = new ArrayList<>(); List batchUids = new ArrayList<>(); do { RecordData data = extractRecordData(resultSet); // 👈 抽取数据 batchRecords.add(data); batchUids.add(data.uid); } while (resultSet.next()); logger.info("本批次共 {} 条记录", batchRecords.size()); // 👇 步骤2:批量查询哪些 uid 已存在(性能优化) Set existingUids = getExistingUids(mysqlConn, batchUids); logger.info("已存在记录数: {}", existingUids.size()); // 👇 步骤3:准备批量插入 try (PreparedStatement mysqlStmt = mysqlConn.prepareStatement(insertSql)) { for (RecordData data : batchRecords) { if (validFourTypes.contains(data.gtype)) { logger.debug("跳过 validFourTypes 类型记录,gtype={}, uid={}", data.gtype, data.uid); continue; } if ("4".equals(data.operation_platform)) { logger.debug("跳过 operation_platform=4 的记录,uid={}", data.uid); continue; } // 👇 跳过已存在的记录(避免重复更新用户余额) if (existingUids.contains(data.uid)) { logger.debug("跳过重复记录,uid={}", data.uid); continue; } // 👇 设置插入参数 setStatementParams(mysqlStmt, data); mysqlStmt.addBatch(); // 👇 只有新记录才更新用户余额(关键修复!) updateUserBalance(mysqlConn, data); } // 👇 执行批量插入 int[] results = mysqlStmt.executeBatch(); logger.info("成功插入新记录 {} 条", results.length); } offset += pageSize; } } logger.info("✅ 数据同步完成"); } catch (SQLException e) { logger.error("数据库操作失败", e); throw new RuntimeException("同步失败", e); } } static class RecordData { int gtype, jwcode, free, core_jb, buy_jb; Timestamp cz_time; String cz_user, cz_bz, operation_platform, goods_name, uid; String orderNumber; // 预生成,避免重复计算 } private RecordData extractRecordData(ResultSet rs) throws SQLException { RecordData data = new RecordData(); data.gtype = rs.getInt("gtype"); data.jwcode = rs.getInt("jwcode"); data.free = rs.getInt("free"); data.core_jb = rs.getInt("core_jb"); data.buy_jb = rs.getInt("buy_jb"); data.cz_time = rs.getTimestamp("cz_time"); data.cz_user = rs.getString("cz_user"); data.cz_bz = rs.getString("cz_bz"); data.operation_platform = rs.getString("operation_platform"); data.goods_name = rs.getString("goods_name"); data.uid = rs.getString("id"); // 预生成订单号(避免在循环中重复生成) String timestampPart = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS")); String uuidPart = UUID.randomUUID().toString().replaceAll("-", ""); data.orderNumber = timestampPart + "_" + uuidPart; return data; } private Set getExistingUids(Connection conn, List uids) throws SQLException { if (uids.isEmpty()) return Collections.emptySet(); String placeholders = String.join(",", Collections.nCopies(uids.size(), "?")); String sql = "SELECT uid FROM user_gold_record WHERE uid IN (" + placeholders + ")"; Set existing = new HashSet<>(); try (PreparedStatement stmt = conn.prepareStatement(sql)) { for (int i = 0; i < uids.size(); i++) { stmt.setString(i + 1, uids.get(i)); } try (ResultSet rs = stmt.executeQuery()) { while (rs.next()) { existing.add(rs.getString("uid")); } } } return existing; } private void setStatementParams(PreparedStatement stmt, RecordData data) throws SQLException { String name = data.cz_user; // 设置 admin_id if (StringUtils.isNumeric(name)) { try { String adminIdStr = adminService.getId(name); if (adminIdStr != null && StringUtils.isNumeric(adminIdStr)) { stmt.setInt(14, Integer.parseInt(adminIdStr)); } else { stmt.setInt(14, 99999); } } catch (Exception e) { logger.warn("解析 admin_id 失败,cz_user={}", name, e); stmt.setInt(14, 99999); } } else { stmt.setInt(14, 99999); } // refund_type, refund_model stmt.setString(10, null); stmt.setNull(11, Types.INTEGER); // 根据 gtype 设置 type 和 order_code if (validFourTypes.contains(data.gtype)) { throw new IllegalArgumentException("不应处理 validFourTypes 类型,应在上层过滤"); // 安全兜底 } if (validZeroTypes.contains(data.gtype)) { stmt.setInt(13, 0); stmt.setNull(20, 0); stmt.setString(1, "ERPCZ_" + data.orderNumber); } else if (validOneTypes.contains(data.gtype)) { stmt.setInt(13, 1); stmt.setInt(20, 0); stmt.setString(1, "ERPXF_" + data.orderNumber); } else if (validTwoTypes.contains(data.gtype)) { stmt.setInt(13, 2); stmt.setInt(20, 0); stmt.setString(1, "ERPTK_" + data.orderNumber); stmt.setString(10, "退款商品"); stmt.setInt(11, 0); } else if (validThreeTypes.contains(data.gtype)) { stmt.setInt(13, 3); stmt.setNull(20, Types.INTEGER); stmt.setString(1, "ERPQT_" + data.orderNumber); } stmt.setInt(2, data.jwcode); stmt.setInt(3, data.free + data.core_jb + data.buy_jb); stmt.setInt(4, data.buy_jb); if (currentMonth.getValue() >= 7) { stmt.setInt(5, data.free); stmt.setInt(6, 0); } else { stmt.setInt(5, 0); stmt.setInt(6, data.free); } stmt.setInt(7, data.core_jb); // pay_platform String platform = data.operation_platform; if ("1".equals(platform)) { stmt.setString(8, "ERP"); } else if ("2".equals(platform)) { stmt.setString(8, "HomilyLink"); } else if ("3".equals(platform)) { stmt.setString(8, "HomilyChart"); } else if ("4".equals(platform)) { throw new IllegalArgumentException("不应处理 platform=4,应在上层过滤"); } else if ("0".equals(platform)) { stmt.setString(8, "初始化金币"); } else { stmt.setString(8, "其他"); } stmt.setString(9, data.goods_name); stmt.setString(12, data.cz_bz); stmt.setInt(15, 3); stmt.setTimestamp(16, data.cz_time); if (data.cz_bz != null && data.cz_bz.contains("测试") && data.cz_bz.contains("员工")) { stmt.setInt(17, 0); } else { stmt.setInt(17, 1); } stmt.setTimestamp(18, data.cz_time); stmt.setTimestamp(19, data.cz_time); stmt.setString(21, data.uid); } private void updateUserBalance(Connection conn, RecordData data) throws Exception { logger.info("处理用户余额更新,jwcode={}", data.jwcode); User user = userService.selectAllUser(String.valueOf(data.jwcode)); BigDecimal freeBD = BigDecimal.valueOf(data.free); BigDecimal buyJbBD = BigDecimal.valueOf(data.buy_jb); BigDecimal coreJbBD = BigDecimal.valueOf(data.core_jb); if (ObjectUtils.isEmpty(user)) { logger.info("用户不存在,jwcode={}", data.jwcode); user = new User(); String country = "未知"; String name = "未知"; try { BaseDES des = new BaseDES(); String desjwcode = des.encrypt(String.valueOf(data.jwcode)); Map requestBody = new HashMap<>(); requestBody.put("jwcode", desjwcode); HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); HttpEntity> entity = new HttpEntity<>(requestBody, headers); ResponseEntity response = restTemplate.exchange( "http://hwapi.rzfwq.com/hwjnApp/hwhc-login/hwhclogin/hc/login/clent/info", HttpMethod.POST, entity, Map.class); if (response.getStatusCode().is2xxSuccessful()) { Map responseBody = response.getBody(); if (responseBody != null) { Map dataMap = (Map) responseBody.get("data"); if (dataMap != null) { name = (String) dataMap.get("name"); country = getCountryWithDefault(dataMap, "未知"); logger.info("获取用户信息成功: name={}, country={}", name, country); } } } } catch (Exception e) { logger.warn("获取用户信息失败,jwcode={}", data.jwcode, e); country = "未知"; } String market = marketService.getMarketIdsDao(country); user.setJwcode(data.jwcode); user.setName(name); user.setMarket(market); userService.addUser(user); logger.info("用户创建成功,jwcode={}", data.jwcode); // 重新查询确保数据完整 user = userService.selectAllUser(String.valueOf(data.jwcode)); } // 更新当前金币 if (currentMonth.getValue() >= 7) { BigDecimal newFreeJune = user.getCurrentFreeJune().add(freeBD); if (newFreeJune.compareTo(BigDecimal.ZERO) >= 0) { user.setCurrentFreeJune(newFreeJune); } else { BigDecimal remaining = newFreeJune; user.setCurrentFreeJune(BigDecimal.ZERO); user.setCurrentFreeDecember(user.getCurrentFreeDecember().add(remaining)); } } else { BigDecimal newFreeDec = user.getCurrentFreeDecember().add(freeBD); if (newFreeDec.compareTo(BigDecimal.ZERO) >= 0) { user.setCurrentFreeDecember(newFreeDec); } else { BigDecimal remaining = newFreeDec; user.setCurrentFreeDecember(BigDecimal.ZERO); user.setCurrentFreeJune(remaining); } } user.setCurrentPermanentGold(user.getCurrentPermanentGold().add(buyJbBD)); user.setCurrentTaskGold(user.getCurrentTaskGold().add(coreJbBD)); // 更新统计字段 if (validZeroTypes.contains(data.gtype)) { user.setRechargeNum(user.getRechargeNum() + 1); user.setSumPermanentGold(user.getSumPermanentGold().add(buyJbBD)); user.setSumTaskGold(user.getSumTaskGold().add(coreJbBD)); if (currentMonth.getValue() >= 7) { user.setSumFreeJune(user.getSumFreeJune().add(freeBD)); } else { user.setSumFreeDecember(user.getSumFreeDecember().add(freeBD)); } } else if (validOneTypes.contains(data.gtype)) { user.setConsumeNum(user.getConsumeNum() + 1); user.setSumConsumePermanent(user.getSumConsumePermanent().add(buyJbBD)); user.setSumConsumeTask(user.getSumConsumeTask().add(coreJbBD)); user.setSumConsumeFree(user.getSumConsumeFree().add(freeBD)); } userService.updateAllGold(user); logger.info("用户余额更新成功,jwcode={}", data.jwcode); } private boolean checkRecordExists(Connection conn, String uid) throws SQLException { String sql = "SELECT 1 FROM user_gold_record WHERE uid = ? LIMIT 1"; try (PreparedStatement stmt = conn.prepareStatement(sql)) { stmt.setString(1, uid); try (ResultSet rs = stmt.executeQuery()) { return rs.next(); } } } private String getCountryWithDefault(Map dataMap, String defaultValue) { Object countryObj = dataMap.get("treelist"); if (countryObj instanceof String) { String countryStr = ((String) countryObj).trim(); if (countryStr.isEmpty()) { return defaultValue; } String[] parts = countryStr.split("-"); return parts.length >= 3 ? parts[2] : defaultValue; } return defaultValue; } }