|
|
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.LocalDate;import java.time.LocalDateTime;import java.time.Month;import java.time.format.DateTimeFormatter;import java.util.*;import java.util.Date;
@Service
public class MysqlServiceImpl implements MysqlService { @Autowired private RestTemplate restTemplate;
Set<Integer> 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,67)); Set<Integer> validOneTypes = new HashSet<>(Arrays.asList(9, 15, 17, 25, 27, 37, 41, 42, 43, 50, 51, 62,68)); Set<Integer> validTwoTypes = new HashSet<>(Arrays.asList(52,61)); Set<Integer> validThreeTypes = new HashSet<>(Arrays.asList(10, 16, 30, 31, 32, 33, 34, 39, 44)); Set<Integer> 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",rollbackFor = Exception.class) // 👈 保证插入和用户更新在一个事务
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,uid 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,link_id) 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<RecordData> batchRecords = new ArrayList<>(); List<String> 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<String> 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); } }
@Override @Transactional(transactionManager = "mysqlTransactionManager") // 👈 保证插入和用户更新在一个事务
public void getSqlserverDataDay() 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, uid 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,link_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """;
while (hasMoreData) { try (PreparedStatement sqlServerStmt = sqlServerConn.prepareStatement(querySql)) { // 👇 恢复动态时间参数
sqlServerStmt.setTimestamp(1, Timestamp.valueOf(LocalDate.now().minusDays(1).atStartOfDay())); sqlServerStmt.setInt(2, offset); sqlServerStmt.setInt(3, pageSize);
ResultSet resultSet = sqlServerStmt.executeQuery();
if (!resultSet.next()) { logger.info("无更多数据，offset={}", offset); break; }
// 👇 步骤1：收集本批次所有记录
List<RecordData> batchRecords = new ArrayList<>(); List<String> 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<String> 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; // 预生成，避免重复计算
String linkId; }
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"); data.linkId = rs.getString("uid");
// 预生成订单号（避免在循环中重复生成）
String timestampPart = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS")); String uuidPart = UUID.randomUUID().toString().replaceAll("-", ""); data.orderNumber = timestampPart + "_" + uuidPart;
return data; }
private Set<String> getExistingUids(Connection conn, List<String> 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<String> 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.setInt(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 if ("5".equals(platform)){ stmt.setString(8, "DC"); }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); } Timestamp now = new Timestamp(System.currentTimeMillis()); stmt.setTimestamp(18, now); stmt.setTimestamp(19, data.cz_time); stmt.setString(21, data.uid); stmt.setString(22, data.linkId); }
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<String, String> requestBody = new HashMap<>(); requestBody.put("jwcode", desjwcode);
HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); HttpEntity<Map<String, String>> entity = new HttpEntity<>(requestBody, headers);
ResponseEntity<Map> 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<String, Object> responseBody = response.getBody(); if (responseBody != null) { Map<String, Object> dataMap = (Map<String, Object>) 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<String, Object> 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; }}
|
