package io.jack.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 數(shù)據(jù)批量入庫(kù)服務(wù)類
 */
@Component
@Slf4j
public class BatchDataStorageService implements InitializingBean {

    /**
     * 最大批次數(shù)量
     */
    @Value("${app.db.maxBatchCount:800}")
    private int maxBatchCount;

    /**
     * 最大線程數(shù)
     */
    @Value("${app.db.maxBatchThreads:100}")
    private int maxBatchThreads;

    /**
     * 超時(shí)時(shí)間，單位毫秒
     */
    @Value("${app.db.batchTimeout:3000}")
    private int batchTimeout;

    /**
     * 當(dāng)前批次數(shù)量
     */
    private int batchCount = 0;

    /**
     * 當(dāng)前批次號(hào)
     */
    private static long batchNo = 0;

    /**
     * 線程池執(zhí)行器
     */
    private ExecutorService executorService = null;

    /**
     * 緩存服務(wù)
     */
    @Resource
    private CacheService cacheService;

    /**
     * 設(shè)備實(shí)時(shí)服務(wù)
     */
    @Resource
    private DeviceRealTimeService deviceRealTimeService;

    /**
     * Redis工具類
     */
    @Resource
    private RedisUtils redisUtils;

    /**
     * 初始化線程池
     */
    @Override
    public void afterPropertiesSet() {
        executorService = Executors.newFixedThreadPool(maxBatchThreads);
    }

    /**
     * 保存設(shè)備實(shí)時(shí)數(shù)據(jù)
     * 
     * @param deviceRealTimeDTO 設(shè)備實(shí)時(shí)數(shù)據(jù)傳輸對(duì)象
     */
    public void saveRealTimeData(DeviceRealTimeDTO deviceRealTimeDTO) {
        final String failedCacheKey = "device:real_time:failed_records";

        try {
            // 生成批次和持續(xù)時(shí)間的緩存鍵
            String durationKey = "device:real_time:batchDuration" + batchNo;
            String batchKey = "device:real_time:batch" + batchNo;

            // 如果當(dāng)前批次持續(xù)時(shí)間不存在，則創(chuàng)建并啟動(dòng)超時(shí)處理線程
            if (!cacheService.exists(durationKey)) {
                cacheService.put(durationKey, System.currentTimeMillis());
                new BatchTimeoutCommitThread(batchKey, durationKey, failedCacheKey).start();
            }

            // 將設(shè)備實(shí)時(shí)數(shù)據(jù)加入當(dāng)前批次
            cacheService.lPush(batchKey, deviceRealTimeDTO);
            if (++batchCount >= maxBatchCount) {
                // 達(dá)到最大批次，執(zhí)行入庫(kù)邏輯
                dataStorage(durationKey, batchKey, failedCacheKey);
            }

        } catch (Exception ex) {
            log.warn("[DB:FAILED] 設(shè)備上報(bào)記錄入批處理集合異常: " + ex.getMessage() + ", DeviceRealTimeDTO: " + JSON.toJSONString(deviceRealTimeDTO), ex);
            cacheService.lPush(failedCacheKey, deviceRealTimeDTO);
        } finally {
            updateRealTimeData(deviceRealTimeDTO);
        }
    }

    /**
     * 更新實(shí)時(shí)數(shù)據(jù)到Redis
     * 
     * @param deviceRealTimeDTO 設(shè)備實(shí)時(shí)數(shù)據(jù)傳輸對(duì)象
     */
    private void updateRealTimeData(DeviceRealTimeDTO deviceRealTimeDTO) {
        redisUtils.set("real_time:" + deviceRealTimeDTO.getDeviceId(), JSONArray.toJSONString(deviceRealTimeDTO));
    }

    /**
     * 批量入庫(kù)處理
     * 
     * @param durationKey 持續(xù)時(shí)間標(biāo)識(shí)
     * @param batchKey    批次標(biāo)識(shí)
     * @param failedCacheKey 錯(cuò)誤記錄標(biāo)識(shí)
     */
    private void dataStorage(String durationKey, String batchKey, String failedCacheKey) {
        batchNo++;
        batchCount = 0;
        cacheService.del(durationKey);
        if (batchNo >= Long.MAX_VALUE) {
            batchNo = 0;
        }
        executorService.execute(new BatchWorker(batchKey, failedCacheKey));
    }

    /**
     * 批量工作線程
     */
    private class BatchWorker implements Runnable {

        private final String failedCacheKey;
        private final String batchKey;

        public BatchWorker(String batchKey, String failedCacheKey) {
            this.batchKey = batchKey;
            this.failedCacheKey = failedCacheKey;
        }

        @Override
        public void run() {
            final List<DeviceRealTimeDTO> deviceRealTimeDTOList = new ArrayList<>();
            try {
                // 從緩存中獲取批量數(shù)據(jù)
                DeviceRealTimeDTO deviceRealTimeDTO = cacheService.lPop(batchKey);
                while (deviceRealTimeDTO != null) {
                    deviceRealTimeDTOList.add(deviceRealTimeDTO);
                    deviceRealTimeDTO = cacheService.lPop(batchKey);
                }

                long timeMillis = System.currentTimeMillis();

                try {
                    // 將DTO轉(zhuǎn)換為實(shí)體對(duì)象并批量入庫(kù)
                    List<DeviceRealTimeEntity> deviceRealTimeEntityList = ConvertUtils.sourceToTarget(deviceRealTimeDTOList, DeviceRealTimeEntity.class);
                    deviceRealTimeService.insertBatch(deviceRealTimeEntityList);
                } finally {
                    cacheService.del(batchKey);
                    log.info("[DB:BATCH_WORKER] 批次：" + batchKey + "，保存設(shè)備上報(bào)記錄數(shù)：" + deviceRealTimeDTOList.size() + ", 耗時(shí)：" + (System.currentTimeMillis() - timeMillis) + "ms");
                }
            } catch (Exception e) {
                log.warn("[DB:FAILED] 設(shè)備上報(bào)記錄批量入庫(kù)失?。? + e.getMessage() + ", DeviceRealTimeDTO: " + deviceRealTimeDTOList.size(), e);
                for (DeviceRealTimeDTO deviceRealTimeDTO : deviceRealTimeDTOList) {
                    cacheService.lPush(failedCacheKey, deviceRealTimeDTO);
                }
            }
        }
    }

    /**
     * 批次超時(shí)提交線程
     */
    class BatchTimeoutCommitThread extends Thread {

        private final String batchKey;
        private final String durationKey;
        private final String failedCacheKey;

        public BatchTimeoutCommitThread(String batchKey, String durationKey, String failedCacheKey) {
            this.batchKey = batchKey;
            this.durationKey = durationKey;
            this.failedCacheKey = failedCacheKey;
            this.setName("batch-thread-" + batchKey);
        }

        @Override
        public void run() {
            try {
                Thread.sleep(batchTimeout);
            } catch (InterruptedException e) {
                log.error("[DB] 內(nèi)部錯(cuò)誤，直接提交：" + e.getMessage());
            }

            if (cacheService.exists(durationKey)) {
                // 達(dá)到最大批次的超時(shí)間，執(zhí)行入庫(kù)邏輯
                dataStorage(durationKey, batchKey, failedCacheKey);
            }
        }
    }
}

package io.jack.service;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 緩存服務(wù)類，提供簡(jiǎn)易的緩存機(jī)制
 */
@Component
public class CacheService implements InitializingBean {

    /**
     * 內(nèi)存緩存，用于存儲(chǔ)數(shù)據(jù)
     */
    private Map<String, Object> objectCache = new ConcurrentHashMap<>();

    /**
     * 統(tǒng)計(jì)緩存，用于統(tǒng)計(jì)計(jì)數(shù)
     */
    private Map<String, AtomicLong> statCache = new ConcurrentHashMap<>();

    /**
     * 初始化統(tǒng)計(jì)緩存
     */
    @Override
    public void afterPropertiesSet() {
        statCache.put("terminals", new AtomicLong(0));
        statCache.put("connections", new AtomicLong(0));
    }

    /**
     * 增加指定統(tǒng)計(jì)項(xiàng)的計(jì)數(shù)
     * 
     * @param statName 統(tǒng)計(jì)項(xiàng)名稱
     * @return 增加后的計(jì)數(shù)值
     */
    public long incr

(String statName) {
        statCache.putIfAbsent(statName, new AtomicLong(0));
        return statCache.get(statName).incrementAndGet();
    }

    /**
     * 減少指定統(tǒng)計(jì)項(xiàng)的計(jì)數(shù)
     * 
     * @param statName 統(tǒng)計(jì)項(xiàng)名稱
     * @return 減少后的計(jì)數(shù)值
     */
    public long decr(String statName) {
        statCache.putIfAbsent(statName, new AtomicLong(0));
        return statCache.get(statName).decrementAndGet();
    }

    /**
     * 獲取指定統(tǒng)計(jì)項(xiàng)的當(dāng)前計(jì)數(shù)值
     * 
     * @param statName 統(tǒng)計(jì)項(xiàng)名稱
     * @return 當(dāng)前計(jì)數(shù)值
     */
    public long stat(String statName) {
        statCache.putIfAbsent(statName, new AtomicLong(0));
        return statCache.get(statName).get();
    }

    /**
     * 存儲(chǔ)數(shù)據(jù)
     * 
     * @param key 緩存鍵
     * @param object 緩存數(shù)據(jù)
     */
    public <T> void put(String key, T object) {
        objectCache.put(key, object);
    }

    /**
     * 獲取數(shù)據(jù)
     * 
     * @param key 緩存鍵
     * @return 緩存數(shù)據(jù)
     */
    public <T> T get(String key) {
        return (T) objectCache.get(key);
    }

    /**
     * 刪除數(shù)據(jù)
     * 
     * @param key 緩存鍵
     */
    public void remove(String key) {
        objectCache.remove(key);
    }

    /**
     * 存儲(chǔ)哈希表數(shù)據(jù)
     * 
     * @param key 哈希表鍵
     * @param subkey 哈希表子鍵
     * @param value 哈希表值
     */
    public void hSet(String key, String subkey, Object value) {
        synchronized (objectCache) {
            Map<String, Object> submap = (Map<String, Object>) objectCache.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
            submap.put(subkey, value);
        }
    }

    /**
     * 獲取哈希表數(shù)據(jù)
     * 
     * @param key 哈希表鍵
     * @param subkey 哈希表子鍵
     * @return 哈希表值
     */
    public <T> T hGet(String key, String subkey) {
        synchronized (objectCache) {
            Map<String, Object> submap = (Map<String, Object>) objectCache.get(key);
            return submap != null ? (T) submap.get(subkey) : null;
        }
    }

    /**
     * 判斷哈希表子鍵是否存在
     * 
     * @param key 哈希表鍵
     * @param subkey 哈希表子鍵
     * @return 是否存在
     */
    public boolean hExists(String key, String subkey) {
        synchronized (objectCache) {
            Map<String, Object> submap = (Map<String, Object>) objectCache.get(key);
            return submap != null && submap.containsKey(subkey);
        }
    }

    /**
     * 將數(shù)據(jù)推入列表
     * 
     * @param key 列表鍵
     * @param value 列表值
     */
    public void lPush(String key, Object value) {
        synchronized (objectCache) {
            LinkedList<Object> queue = (LinkedList<Object>) objectCache.computeIfAbsent(key, k -> new LinkedList<>());
            queue.addLast(value);
        }
    }

    /**
     * 從列表中彈出數(shù)據(jù)
     * 
     * @param key 列表鍵
     * @return 列表值
     */
    public <T> T lPop(String key) {
        synchronized (objectCache) {
            LinkedList<Object> queue = (LinkedList<Object>) objectCache.get(key);
            return queue != null && !queue.isEmpty() ? (T) queue.removeLast() : null;
        }
    }

    /**
     * 刪除緩存數(shù)據(jù)
     * 
     * @param key 緩存鍵
     */
    public void del(String key) {
        objectCache.remove(key);
    }

    /**
     * 判斷緩存鍵是否存在
     * 
     * @param key 緩存鍵
     * @return 是否存在
     */
    public boolean exists(String key) {
        return objectCache.containsKey(key);
    }
}