package com.xzzn.ems.utils;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.xzzn.common.core.redis.RedisCache;
import com.xzzn.common.utils.DateUtils;
import com.xzzn.ems.domain.EmsBatteryData;
import com.xzzn.ems.domain.EmsBatteryDataDay;
import com.xzzn.ems.domain.EmsBatteryDataHour;
import com.xzzn.ems.domain.EmsBatteryDataMonth;
import com.xzzn.ems.mapper.EmsBatteryDataDayMapper;
import com.xzzn.ems.mapper.EmsBatteryDataHourMapper;
import com.xzzn.ems.mapper.EmsBatteryDataMonthMapper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public abstract class AbstractBatteryDataProcessor {

    private static final Log log = LogFactory.getLog(AbstractBatteryDataProcessor.class);

    protected static final DateTimeFormatter HOUR_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH");
    protected static final DateTimeFormatter DAY_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd");
    protected static final DateTimeFormatter MONTH_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM");
    // 公共缓存：<batteryId_granularity, 缓存值>
    protected final Map<String, CacheValue> tempCache = new ConcurrentHashMap<>();
    // 公共锁缓存：<batteryId_granularity, 锁>
    protected final Map<String, ReentrantLock> lockCache = new ConcurrentHashMap<>();
    protected final ObjectMapper objectMapper;

    @Autowired
    private EmsBatteryDataHourMapper emsBatteryDataHourMapper;
    @Autowired
    private EmsBatteryDataDayMapper emsBatteryDataDayMapper;
    @Autowired
    private EmsBatteryDataMonthMapper emsBatteryDataMonthMapper;
    @Autowired
    private RedisCache redisCache;

    // 公共线程池（IO密集型，子类共享）
    protected ThreadPoolTaskExecutor ioExecutor;
    // 批量缓冲区（线程安全队列，按级别区分）
    protected ConcurrentLinkedQueue<EmsBatteryData> hourBatchQueue = new ConcurrentLinkedQueue<>();
    protected ConcurrentLinkedQueue<EmsBatteryData> dayBatchQueue = new ConcurrentLinkedQueue<>();
    protected ConcurrentLinkedQueue<EmsBatteryData> monthBatchQueue = new ConcurrentLinkedQueue<>();
    // 批量提交阈值（满100条触发数据库操作）
    protected static final int BATCH_THRESHOLD = 50;
    // 死锁重试等待时间基数（毫秒）
    private static final int RETRY_BASE_DELAY = 100;
    // 随机数生成器，用于重试退避
    private static final Random random = new Random();

    // 初始化方法（构造时初始化线程池）
    public AbstractBatteryDataProcessor(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
        ioExecutor = new ThreadPoolTaskExecutor();
        int coreThreads = Runtime.getRuntime().availableProcessors() * 2;
        ioExecutor.setCorePoolSize(coreThreads);
        ioExecutor.setMaxPoolSize(coreThreads * 2);
        ioExecutor.setQueueCapacity(10000);
        ioExecutor.setThreadNamePrefix("battery-io-");
        // 线程池饱和策略：使用调用者线程执行，避免任务丢失
        ioExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        ioExecutor.initialize();
    }

    // 模板方法：定义数据处理主流程（子类无需重写）
    public  void processBatch(List<EmsBatteryData> batchData) {
        // 1. 数据分片（实现通用分片逻辑）
        List<List<EmsBatteryData>> shards = splitIntoShards(batchData, 50);
        int totalShards = shards.size();

        // 2. 并行处理每个分片（使用父类线程池）
        for (int i = 0; i < totalShards; i++) {
            List<EmsBatteryData> shard = shards.get(i);
            boolean isLastShard = (i == totalShards - 1); // 判断是否为最后一个分片

            // 传递分片索引和是否最后一个分片的标记
            ioExecutor.execute(() -> processShard(shard, isLastShard));
        }
    }

    /**
     * 将大批量数据拆分为多个小分片
     * @param dataList 原始数据列表
     * @param shardSize 每个分片的最大大小（如100条/片）
     * @return 分片后的列表（每个元素是一个子列表）
     */
    protected List<List<EmsBatteryData>> splitIntoShards(List<EmsBatteryData> dataList, int shardSize) {
        // 1. 校验参数：避免空列表或无效分片大小
        if (dataList == null || dataList.isEmpty()) {
            return new ArrayList<>();
        }
        if (shardSize <= 0) {
            throw new IllegalArgumentException("分片大小必须大于0");
        }

        // 2. 计算总分片数：向上取整（如150条数据，分片大小100 → 2片）
        int totalSize = dataList.size();
        int shardCount = (totalSize + shardSize - 1) / shardSize;

        // 3. 拆分数据为分片（使用IntStream生成分片索引）
        return IntStream.range(0, shardCount)
                .mapToObj(shardIndex -> {
                    // 计算当前分片的起始索引和结束索引
                    int startIndex = shardIndex * shardSize;
                    int endIndex = Math.min(startIndex + shardSize, totalSize);
                    // 截取子列表作为一个分片
                    return dataList.subList(startIndex, endIndex);
                })
                .collect(Collectors.toList());
    }

    // 处理单个分片（父类定义流程，调用子类实现的具体判断和更新逻辑）
    private void processShard(List<EmsBatteryData> shard, boolean isLastShard) {
        int shardSize = shard.size();
        log.info("分片校验：" + shardSize + "<UNK>");
        for (int i = 0; i < shardSize; i++) {
            EmsBatteryData data = shard.get(i);
            if (!isValidData(data)) continue;

            // 判断当前数据是否为最后一个分片的最后一条数据
            boolean isLastData = isLastShard && (i == shardSize - 1);
            // shouldUpdate判断是否需要更新
            if (shouldUpdate(data, "hour")) {
                addToHourBatch(data, isLastData); // 加入小时级批量队列
            }
            if (shouldUpdate(data, "day")) {
                addToDayBatch(data, isLastData); // 加入天级批量队列
            }
            if (shouldUpdate(data, "month")) {
                addToMonthBatch(data, isLastData);
            }
        }
    }

    // 加入小时级批量队列，达到阈值时触发批量更新
    protected void addToHourBatch(EmsBatteryData data, boolean isLastData) {
        hourBatchQueue.add(data);
        if (hourBatchQueue.size() >= BATCH_THRESHOLD || isLastData) {
            List<EmsBatteryData> batch = drainQueue(hourBatchQueue);
            ioExecutor.execute(() -> batchUpdateHour(batch)); // 异步批量更新
        }
    }

    // 加入天级批量队列，达到阈值时触发批量更新
    protected void addToDayBatch(EmsBatteryData data, boolean isLastData) {
        dayBatchQueue.add(data);
        if (dayBatchQueue.size() >= BATCH_THRESHOLD || isLastData) {
            List<EmsBatteryData> batch = drainQueue(dayBatchQueue);
            ioExecutor.execute(() -> batchUpdateDay(batch)); // 异步批量更新
        }
    }

    // 加入月级批量队列，达到阈值时触发批量更新
    protected void addToMonthBatch(EmsBatteryData data, boolean isLastData) {
        monthBatchQueue.add(data);
        if (monthBatchQueue.size() >= BATCH_THRESHOLD || isLastData) {
            List<EmsBatteryData> batch = drainQueue(monthBatchQueue);
            ioExecutor.execute(() -> batchUpdateMonth(batch)); // 异步批量更新
        }
    }

    // 实现父类的抽象方法：批量更新小时级数据
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void batchUpdateHour(List<EmsBatteryData> batch) {
        if (batch.isEmpty()) return;

        int maxRetry = 3; // 最多重试3次
        int retryCount = 0;
        while (retryCount < maxRetry) {
            try {
                // 关键优化：按唯一键排序，保证所有线程处理相同记录时顺序一致
                List<EmsBatteryData> sortedBatch = batch.stream()
                        .sorted(Comparator.comparing(data -> generateCacheKey(data, "hour"))) // 用唯一键排序
                        .collect(Collectors.toList());

                // 1. 转换原始数据为小时级统计对象（分组取最高温）
                List<EmsBatteryDataHour> hourDataList = sortedBatch.stream()
                        .collect(Collectors.groupingBy(
                                data -> generateCacheKey(data, "hour"), // 分组键
                                Collectors.collectingAndThen(
                                        Collectors.maxBy((d1, d2) -> d1.getTemperature().compareTo(d2.getTemperature())),
                                        maxData -> convertToHourData(maxData.get()) // 取每组最高温转换为HourData
                                )
                        ))
                        .values()
                        .stream()
                        .sorted(Comparator.comparing(this::getHourDataUniqueKey))
                        .collect(Collectors.toList());

                // 2. 批量插入或更新数据库（依赖数据库唯一索引）
                emsBatteryDataHourMapper.batchInsertOrUpdate(hourDataList);

                // 3. 同步更新Redis缓存（批量操作）
                batchUpdateHourCache(hourDataList);
                log.info("小时级批量更新成功，批次大小：" + batch.size() + "，重试次数：" + retryCount);
                return;
            } catch (Exception e) {
                log.error("小时级批量更新失败", e);
                // 使用改进的死锁判断方法，直接传入异常对象
                if (isDeadlockException(e) && retryCount < maxRetry - 1) {
                    retryCount++;
                    long sleepTime = calculateBackoffTime(retryCount);
                    log.info("检测到死锁，第" + retryCount + "次重试（当前批次大小：" + batch.size() + "）", e);
                    try {
                        // 指数退避策略：100ms, 200ms, 300ms...
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt(); // 保留中断状态
                        break;
                    }
                } else {
                    log.error("小时级批量更新失败（已达最大重试次数）", e);
                    //retryQueue.addAll(batch); // 加入重试队列
                    return;
                }
            }
        }
    }
    // 辅助方法：获取小时级数据的唯一键（用于排序）
    private String getHourDataUniqueKey(EmsBatteryDataHour data) {
        return data.getSiteId() + "_" + data.getBatteryPack() + "_" + data.getClusterDeviceId()
                + "_" + data.getDeviceId() + "_" + data.getHourTime();
    }
    // 批量更新小时级数据缓存-小时级统计数据列表（已按唯一键分组并取最高温）
    public void batchUpdateHourCache(List<EmsBatteryDataHour> hourDataList) {
        if (hourDataList.isEmpty()) return;

        Map<String, String> tempCacheMap = buildCacheMap(hourDataList, "hour");

        try {
            // 批量设置缓存
            redisCache.multiSetWithExpire(tempCacheMap, 1, TimeUnit.HOURS);
        } catch (Exception e) {
            log.error("批量更新小时级缓存失败", e);
            // 缓存更新失败不影响主流程，但需记录日志便于排查
        }
    }
    // 转换为小时级数据对象
    private EmsBatteryDataHour convertToHourData(EmsBatteryData data) {
        LocalDateTime localDateTime = data.getDataTimestamp().toInstant().atZone(ZoneId.of("Asia/Shanghai"))
                .toLocalDateTime();
        LocalDateTime hourStart = localDateTime.truncatedTo(ChronoUnit.HOURS);

        EmsBatteryDataHour hourData = new EmsBatteryDataHour();
        BeanUtils.copyProperties(data, hourData);
        hourData.setUpdateTime(new Date());
        hourData.setHourTime(DateUtils.convertToDate(hourStart));
        return hourData;
    }

    // 实现父类的抽象方法：批量更新天级数据
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void batchUpdateDay(List<EmsBatteryData> batch) {
        if (batch.isEmpty()) return;

        int maxRetry = 3; // 最多重试3次
        int retryCount = 0;
        while (retryCount < maxRetry) {
            try {
                // 关键优化：按唯一键排序，保证所有线程处理相同记录时顺序一致
                List<EmsBatteryData> sortedBatch = batch.stream()
                        .sorted(Comparator.comparing(data -> generateCacheKey(data, "day"))) // 用唯一键排序
                        .collect(Collectors.toList());

                // 1. 转换原始数据为天级统计对象（分组取最高温）
                List<EmsBatteryDataDay> dayDataList = sortedBatch.stream()
                        .collect(Collectors.groupingBy(
                                data -> generateCacheKey(data, "day"), // 分组键
                                Collectors.collectingAndThen(
                                        Collectors.maxBy((d1, d2) -> d1.getTemperature().compareTo(d2.getTemperature())),
                                        maxData -> convertToDayData(maxData.get()) // 取每组最高温转换为HourData
                                )
                        ))
                        .values()
                        .stream()
                        .sorted(Comparator.comparing(this::getDayDataUniqueKey))
                        .collect(Collectors.toList());

                // 2. 批量插入或更新数据库（依赖数据库唯一索引）
                emsBatteryDataDayMapper.batchInsertOrUpdate(dayDataList);

                // 3. 同步更新Redis缓存（批量操作）
                batchUpdateDayCache(dayDataList);
                log.info("天级批量更新成功，批次大小：" + batch.size() + "，重试次数：" + retryCount);
                return;
            } catch (Exception e) {
                log.error("天级批量更新失败", e);
                // 使用改进的死锁判断方法，直接传入异常对象
                if (isDeadlockException(e) && retryCount < maxRetry - 1) {
                    retryCount++;
                    long sleepTime = calculateBackoffTime(retryCount);
                    log.info("检测到死锁，第" + retryCount + "次重试（当前批次大小：" + batch.size() + "）", e);
                    try {
                        // 指数退避策略：100ms, 200ms, 300ms...
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt(); // 保留中断状态
                        break;
                    }
                } else {
                    log.error("天级批量更新失败（已达最大重试次数）", e);
                    //retryQueue.addAll(batch); // 加入重试队列
                    return;
                }
            }
        }
    }
    // 辅助方法：获取天级数据的唯一键（用于排序）
    private String getDayDataUniqueKey(EmsBatteryDataDay data) {
        return data.getSiteId() + "_" + data.getBatteryPack() + "_" + data.getClusterDeviceId()
                + "_" + data.getDeviceId() + "_" + data.getDayTime();
    }
    // 批量更新天级数据缓存-天级统计数据列表（已按唯一键分组并取最高温）
    public void batchUpdateDayCache(List<EmsBatteryDataDay> dayDataList) {
        if (dayDataList.isEmpty()) return;

        Map<String, String> tempCacheMap = buildCacheMap(dayDataList, "hour");

        try {
            // 批量设置缓存
            redisCache.multiSetWithExpire(tempCacheMap, 24, TimeUnit.HOURS);
        } catch (Exception e) {
            log.error("批量更新小时级缓存失败", e);
            // 缓存更新失败不影响主流程，但需记录日志便于排查
        }
    }
    // 转换为天级数据对象
    private EmsBatteryDataDay convertToDayData(EmsBatteryData data) {
        LocalDateTime localDateTime = data.getDataTimestamp().toInstant().atZone(ZoneId.of("Asia/Shanghai"))
                .toLocalDateTime();
        LocalDateTime dayStart = localDateTime.truncatedTo(ChronoUnit.DAYS);

        EmsBatteryDataDay dayData = new EmsBatteryDataDay();
        BeanUtils.copyProperties(data, dayData);
        dayData.setUpdateTime(new Date());
        dayData.setDayTime(DateUtils.convertToDate(dayStart));
        return dayData;
    }

    // 实现父类的抽象方法：批量更新月级数据
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void batchUpdateMonth(List<EmsBatteryData> batch) {
        if (batch.isEmpty()) return;

        int maxRetry = 3; // 最多重试3次
        int retryCount = 0;
        while (retryCount < maxRetry) {
            try {
                // 关键优化：按唯一键排序，保证所有线程处理相同记录时顺序一致
                List<EmsBatteryData> sortedBatch = batch.stream()
                        .sorted(Comparator.comparing(data -> generateCacheKey(data, "month"))) // 用唯一键排序
                        .collect(Collectors.toList());

                // 1. 转换原始数据为小时级统计对象（分组取最高温）
                List<EmsBatteryDataMonth> monthDataList = sortedBatch.stream()
                        .collect(Collectors.groupingBy(
                                data -> generateCacheKey(data, "month"), // 分组键
                                Collectors.collectingAndThen(
                                        Collectors.maxBy((d1, d2) -> d1.getTemperature().compareTo(d2.getTemperature())),
                                        maxData -> convertToMonthData(maxData.get()) // 取每组最高温转换为HourData
                                )
                        ))
                        .values()
                        .stream()
                        .sorted(Comparator.comparing(this::getMonthDataUniqueKey))
                        .collect(Collectors.toList());

                // 2. 批量插入或更新数据库（依赖数据库唯一索引）
                emsBatteryDataMonthMapper.batchInsertOrUpdate(monthDataList);

                // 3. 同步更新Redis缓存（批量操作）
                batchUpdateMonthCache(monthDataList);
                log.info("月级批量更新成功，批次大小：" + batch.size() + "，重试次数：" + retryCount);
                return;
            } catch (Exception e) {
                log.error("月级批量更新失败", e);
                // 使用改进的死锁判断方法，直接传入异常对象
                if (isDeadlockException(e) && retryCount < maxRetry - 1) {
                    retryCount++;
                    long sleepTime = calculateBackoffTime(retryCount);
                    log.info("检测到死锁，第" + retryCount + "次重试（当前批次大小：" + batch.size() + "）", e);
                    try {
                        // 指数退避策略：100ms, 200ms, 300ms...
                        Thread.sleep(sleepTime);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt(); // 保留中断状态
                        break;
                    }
                } else {
                    log.error("月级批量更新失败（已达最大重试次数）", e);
                    //retryQueue.addAll(batch); // 加入重试队列
                    return;
                }
            }
        }
    }
    // 辅助方法：获取月级数据的唯一键（用于排序）
    private String getMonthDataUniqueKey(EmsBatteryDataMonth data) {
        return data.getSiteId() + "_" + data.getBatteryPack() + "_" + data.getClusterDeviceId()
                + "_" + data.getDeviceId() + "_" + data.getMonthTime();
    }
    // 转换为月级数据对象
    private EmsBatteryDataMonth convertToMonthData(EmsBatteryData data) {
        LocalDateTime localDateTime = data.getDataTimestamp().toInstant().atZone(ZoneId.of("Asia/Shanghai"))
                .toLocalDateTime();
        LocalDateTime monthStart = localDateTime.withDayOfMonth(1).withHour(0).withMinute(0)
                .withSecond(0).withNano(0);

        EmsBatteryDataMonth monthData = new EmsBatteryDataMonth();
        BeanUtils.copyProperties(data, monthData);
        monthData.setUpdateTime(new Date());
        monthData.setMonthTime(DateUtils.convertToDate(monthStart));
        return monthData;
    }
    // 批量更新月级数据缓存-月级统计数据列表（已按唯一键分组并取最高温）
    public void batchUpdateMonthCache(List<EmsBatteryDataMonth> monthDataList) {
        if (monthDataList.isEmpty()) return;

        Map<String, String> tempCacheMap = buildCacheMap(monthDataList, "hour");

        try {
            // 批量设置缓存
            redisCache.multiSetWithExpire(tempCacheMap, 24*30, TimeUnit.HOURS);
        } catch (Exception e) {
            log.error("批量更新月级缓存失败", e);
            // 缓存更新失败不影响主流程，但需记录日志便于排查
        }
    }


    /**
     * 校验数据有效性（公共逻辑：检查核心字段）
     */
    protected boolean isValidData(EmsBatteryData data) {
        return data.getDeviceId() != null
                && data.getTemperature() != null
                && data.getDataTimestamp() != null;
    }

    /**
     * 判断是否需要更新（公共缓存逻辑）
     */
    protected boolean shouldUpdate(EmsBatteryData data, String granularity) {
        String cacheKey = generateCacheKey(data, granularity);
        if (cacheKey == null) return true;

        BigDecimal newTemp = data.getTemperature();
        CacheValue localCache = tempCache.get(cacheKey);
        if (localCache != null) {
            return newTemp.compareTo(localCache.maxTemp) > 0;
        }

        //  本地缓存未命中：查Redis
        try {
            String redisValue = redisCache.getCacheObject(cacheKey);
            if (redisValue != null) {
                // Redis有值：转换为BigDecimal比较
                BigDecimal redisMaxTemp = new BigDecimal(redisValue);
                // 同时更新本地缓存（减少下次查询Redis的开销）
                tempCache.put(cacheKey, new CacheValue(redisMaxTemp, LocalDateTime.now()));
                return newTemp.compareTo(redisMaxTemp) > 0;
            }
        } catch (Exception e) {
            log.error("查询Redis缓存失败（key: " + cacheKey + "）", e);
         }


        // 数据库校验（缓存都未命中或新温度更高时）
        BigDecimal dbMaxTemp = queryDbMaxTemp(data, granularity);
        if (dbMaxTemp != null && newTemp.compareTo(dbMaxTemp) <= 0) {
            // 数据库已有更高温度，更新缓存后返回
            CacheValue dbCache = new CacheValue(dbMaxTemp, LocalDateTime.now());
            tempCache.put(cacheKey, dbCache);
            redisCache.setCacheObject(cacheKey, dbMaxTemp.toString()); // 同步到Redis
            return false;
        }

        return true;
    }

    /**
     * 查询数据库中该设备在指定时间粒度下的最大温度
     */
    private BigDecimal queryDbMaxTemp(EmsBatteryData data, String granularity) {
        try {
            LocalDateTime dataTime = data.getDataTimestamp().toInstant()
                    .atZone(ZoneId.of("Asia/Shanghai")).toLocalDateTime();

            String siteId = data.getSiteId();
            String batteryPack = data.getBatteryPack();
            String clusterId = data.getClusterDeviceId();
            String deviceId = data.getDeviceId();

            // 根据粒度查询对应的数据表
            switch (granularity) {
                case "hour":
                    LocalDateTime hourStart = dataTime.truncatedTo(ChronoUnit.HOURS);
                    // 调用Mapper查询该小时的最大温度（需确保Mapper方法存在）
                    EmsBatteryDataHour hourMax = emsBatteryDataHourMapper.findHourMaxTemp(
                            siteId, batteryPack, clusterId, deviceId, hourStart
                    );
                    return hourMax != null ? hourMax.getTemperature() : null;

                case "day":
                    LocalDateTime dayStart = dataTime.truncatedTo(ChronoUnit.DAYS);
                    EmsBatteryDataDay dayMax = emsBatteryDataDayMapper.findDayMaxTemp(
                            siteId, batteryPack, clusterId, deviceId, dayStart
                    );
                    return dayMax != null ? dayMax.getTemperature() : null;

                case "month":
                    // 设为当月1号 与其他表统一格式
                    LocalDateTime monthStart = dataTime.withDayOfMonth(1).withHour(0).withMinute(0)
                            .withSecond(0).withNano(0);
                    EmsBatteryDataMonth monthMax = emsBatteryDataMonthMapper.findMonthMaxTemp(
                            siteId, batteryPack, clusterId, deviceId, monthStart
                    );
                    return monthMax != null ? monthMax.getTemperature() : null;

                default:
                    log.error("不支持的查询粒度：" + granularity);
                    return null;
            }
        } catch (Exception e) {
            log.error("查询数据库最大温度失败", e);
            return null; // 数据库查询失败时，暂不阻止更新（避免业务中断）
        }
    }

    // 生成缓存键（不同级别格式不同）
    private String generateCacheKey(EmsBatteryData data, String granularity) {
        String siteId = data.getSiteId();
        String stackId = data.getBatteryPack();
        String clusterId = data.getClusterDeviceId();
        String batteryId = data.getDeviceId();
        try {
            LocalDateTime localDateTime = data.getDataTimestamp().toInstant().atZone(ZoneId.of("Asia/Shanghai"))
                    .toLocalDateTime();
            String timeStr;
            // 根据级别格式化时间（小时/天/月）
            switch (granularity) {
                case "hour":
                    timeStr = localDateTime.format(HOUR_FORMATTER);
                    break;
                case "day":
                    timeStr = localDateTime.format(DAY_FORMATTER);
                    break;
                case "month":
                    timeStr = localDateTime.format(MONTH_FORMATTER);
                    break;
                default:
                    log.info("不支持的级别：" + granularity);
                    return null;
            }
            // 缓存键格式
            return siteId + "_" + stackId + "_" + clusterId + "_" + batteryId + "_" +timeStr;
        } catch (Exception e) {
            log.error("生成缓存键失败", e);
            return null;
        }
    }

    // 工具方法：从队列中取出所有元素
    protected <T> List<T> drainQueue(ConcurrentLinkedQueue<T> queue) {
        List<T> result = new ArrayList<>();
        T element;
        while ((element = queue.poll()) != null) {
            result.add(element);
        }
        return result;
    }

    /**
     * 获取电池+粒度的专属锁（公共锁逻辑）
     */
    protected ReentrantLock getLock(String siteId, String stackId, String clusterId,
                                    String batteryId, String granularity) {
        String lockKey = siteId + "_" + stackId + "_" + clusterId + "_" + batteryId + "_" + granularity;
        return lockCache.computeIfAbsent(lockKey, k -> new ReentrantLock());
    }

    // 计算退避时间：指数退避 + 随机抖动，避免重试风暴
    private long calculateBackoffTime(int retryCount) {
        // 100ms * 2^retry + 随机0-100ms
        return (long)(RETRY_BASE_DELAY * Math.pow(2, retryCount)) + random.nextInt(100);
    }

    // 判断是否为死锁异常
    private boolean isDeadlockException(Throwable e) {
        // 遍历所有异常链（包括cause的cause...）
        while (e != null) {
            // 检查异常消息是否包含死锁关键字
            if (e.getMessage() != null && e.getMessage().contains("Deadlock found when trying to get lock")) {
                return true;
            }
            // 深入下一层异常
            e = e.getCause();
        }
        return false;
    }

    /**
     * 公共缓存实体（所有站点通用）
     */
    protected static class CacheValue {
        BigDecimal maxTemp;
        LocalDateTime lastUpdateTime;

        public CacheValue(BigDecimal maxTemp, LocalDateTime lastUpdateTime) {
            this.maxTemp = maxTemp;
            this.lastUpdateTime = lastUpdateTime;
        }
    }

    // 通用构建缓存键值对的方法（根据不同数据类型适配）
    private <T> Map<String, String> buildCacheMap(List<T> dataList, String granularity) {
        Map<String, String> cacheMap = new HashMap<>(dataList.size());
        for (T data : dataList) {
            EmsBatteryData cacheData = new EmsBatteryData();
            BeanUtils.copyProperties(data, cacheData);
            String cacheKey = generateCacheKey(cacheData, granularity);

            // 根据数据类型获取温度值（实际项目中可通过接口或反射统一处理）
            String temperature;
            if (data instanceof EmsBatteryDataHour) {
                temperature = ((EmsBatteryDataHour) data).getTemperature().toString();
            } else if (data instanceof EmsBatteryDataDay) {
                temperature = ((EmsBatteryDataDay) data).getTemperature().toString();
            } else if (data instanceof EmsBatteryDataMonth) {
                temperature = ((EmsBatteryDataMonth) data).getTemperature().toString();
            } else {
                throw new IllegalArgumentException("不支持的数据类型");
            }
            cacheMap.put(cacheKey, temperature);
        }
        return cacheMap;
    }

    // 新增定时任务，每1分钟强制刷新一次队列（防止数据长期积压）
    @Scheduled(fixedRate = 60000) // 1分钟 = 60000毫秒
    protected void scheduledFlushBatch() {
        if (!hourBatchQueue.isEmpty()) {
            List<EmsBatteryData> remaining = drainQueue(hourBatchQueue);
            ioExecutor.execute(() -> batchUpdateHour(remaining));
            log.info("定时任务触发，处理小时级剩余数据，数量: " + remaining.size());
        }
        if (!dayBatchQueue.isEmpty()) {
            List<EmsBatteryData> remaining = drainQueue(dayBatchQueue);
            ioExecutor.execute(() -> batchUpdateDay(remaining));
            log.info("定时任务触发，处理天级剩余数据，数量: " + remaining.size());
        }
        if (!monthBatchQueue.isEmpty()) {
            List<EmsBatteryData> remaining = drainQueue(monthBatchQueue);
            ioExecutor.execute(() -> batchUpdateMonth(remaining));
            log.info("定时任务触发，处理月级剩余数据，数量: " + remaining.size());
        }
    }
}
