How Java uses thread pool and Redis to achieve efficient data entry

package ;

import ;
import ;
import .slf4j.Slf4j;
import ;
import ;
import ;

import ;
import ;
import ;
import ;
import ;

/**
  * Data batch storage service category
  */
@Component
@Slf4j
public class BatchDataStorageService implements InitializingBean {

    /**
      * Maximum batch quantity
      */
    @Value("${:800}")
    private int maxBatchCount;

    /**
      * Maximum number of threads
      */
    @Value("${:100}")
    private int maxBatchThreads;

    /**
      * Timeout time, unit milliseconds
      */
    @Value("${:3000}")
    private int batchTimeout;

    /**
      * Current batch quantity
      */
    private int batchCount = 0;

    /**
      * Current batch number
      */
    private static long batchNo = 0;

    /**
      * Thread pool executor
      */
    private ExecutorService executorService = null;

    /**
      * Cache service
      */
    @Resource
    private CacheService cacheService;

    /**
      * Real-time equipment service
      */
    @Resource
    private DeviceRealTimeService deviceRealTimeService;

    /**
      * Redis tool class
      */
    @Resource
    private RedisUtils redisUtils;

    /**
      * Initialize thread pool
      */
    @Override
    public void afterPropertiesSet() {
        executorService = (maxBatchThreads);
    }

    /**
      * Save real-time data of the device
      *
      * @param deviceRealTimeDTO device real-time data transfer object
      */
    public void saveRealTimeData(DeviceRealTimeDTO deviceRealTimeDTO) {
        final String failedCacheKey = "device:real_time:failed_records";

        try {
            // Generate batch and duration cache keys            String durationKey = "device:real_time:batchDuration" + batchNo;
            String batchKey = "device:real_time:batch" + batchNo;

            // If the current batch duration does not exist, create and start the timeout processing thread            if (!(durationKey)) {
                (durationKey, ());
                new BatchTimeoutCommitThread(batchKey, durationKey, failedCacheKey).start();
            }

            // Add real-time data of the device to the current batch            (batchKey, deviceRealTimeDTO);
            if (++batchCount >= maxBatchCount) {
                // To reach the maximum batch, execute the inbound logic                dataStorage(durationKey, batchKey, failedCacheKey);
            }

        } catch (Exception ex) {
            ("[DB:FAILED] The device reports an exception to record the batch collection: " + () + ", DeviceRealTimeDTO: " + (deviceRealTimeDTO), ex);
            (failedCacheKey, deviceRealTimeDTO);
        } finally {
            updateRealTimeData(deviceRealTimeDTO);
        }
    }

    /**
      * Update real-time data to Redis
      *
      * @param deviceRealTimeDTO device real-time data transfer object
      */
    private void updateRealTimeData(DeviceRealTimeDTO deviceRealTimeDTO) {
        ("real_time:" + (), (deviceRealTimeDTO));
    }

    /**
      * Batch storage processing
      *
      * @param durationKey Duration mark
      * @param batchKey batch identification
      * @param failedCacheKey Error record identification
      */
    private void dataStorage(String durationKey, String batchKey, String failedCacheKey) {
        batchNo++;
        batchCount = 0;
        (durationKey);
        if (batchNo >= Long.MAX_VALUE) {
            batchNo = 0;
        }
        (new BatchWorker(batchKey, failedCacheKey));
    }

    /**
      * Batch worker threads
      */
    private class BatchWorker implements Runnable {

        private final String failedCacheKey;
        private final String batchKey;

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

        @Override
        public void run() {
            final List<DeviceRealTimeDTO> deviceRealTimeDTOList = new ArrayList<>();
            try {
                // Get batch data from cache                DeviceRealTimeDTO deviceRealTimeDTO = (batchKey);
                while (deviceRealTimeDTO != null) {
                    (deviceRealTimeDTO);
                    deviceRealTimeDTO = (batchKey);
                }

                long timeMillis = ();

                try {
                    // Convert DTO to entity objects and batch into the library                    List<DeviceRealTimeEntity> deviceRealTimeEntityList = (deviceRealTimeDTOList, );
                    (deviceRealTimeEntityList);
                } finally {
                    (batchKey);
                    ("[DB:BATCH_WORKER] Batch:" + batchKey + ", save the number of records reported by the device:" + () + ", time consuming:" + (() - timeMillis) + "ms");
                }
            } catch (Exception e) {
                ("[DB:FAILED] The device reporting record failed to enter the batch of database:" + () + ", DeviceRealTimeDTO: " + (), e);
                for (DeviceRealTimeDTO deviceRealTimeDTO : deviceRealTimeDTOList) {
                    (failedCacheKey, deviceRealTimeDTO);
                }
            }
        }
    }

    /**
      * Batch timeout submission thread
      */
    class BatchTimeoutCommitThread extends Thread {

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

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

        @Override
        public void run() {
            try {
                (batchTimeout);
            } catch (InterruptedException e) {
                ("[DB] Internal error, submit directly:" + ());
            }

            if ((durationKey)) {
                // The maximum batch timeout is reached, and the inbound logic is executed                dataStorage(durationKey, batchKey, failedCacheKey);
            }
        }
    }
}

package ;

import ;
import ;

import ;
import ;
import ;
import ;

/**
  * Cache service class, providing a simple caching mechanism
  */
@Component
public class CacheService implements InitializingBean {

    /**
      * Memory cache, used to store data
      */
    private Map<String, Object> objectCache = new ConcurrentHashMap<>();

    /**
      * Statistical cache, used for counting
      */
    private Map<String, AtomicLong> statCache = new ConcurrentHashMap<>();

    /**
      * Initialize the statistics cache
      */
    @Override
    public void afterPropertiesSet() {
        ("terminals", new AtomicLong(0));
        ("connections", new AtomicLong(0));
    }

    /**
      * Increase the count of specified statistics
      *
      * @param statName Statistics name
      * @return The incremented count value
      */
    public long incr

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

    /**
      * Reduce the count of specified statistics
      *
      * @param statName Statistics name
      * @return The reduced count value
      */
    public long decr(String statName) {
        (statName, new AtomicLong(0));
        return (statName).decrementAndGet();
    }

    /**
      * Get the current count value of the specified statistics
      *
      * @param statName Statistics name
      * @return Current count value
      */
    public long stat(String statName) {
        (statName, new AtomicLong(0));
        return (statName).get();
    }

    /**
      * Store data
      *
      * @param key cache key
      * @param object caches data
      */
    public <T> void put(String key, T object) {
        (key, object);
    }

    /**
      * Get data
      *
      * @param key cache key
      * @return Cache data
      */
    public <T> T get(String key) {
        return (T) (key);
    }

    /**
      * Delete data
      *
      * @param key cache key
      */
    public void remove(String key) {
        (key);
    }

    /**
      * Store hash table data
      *
      * @param key hash table key
      * @param subkey hash table subkey
      * @param value hash table value
      */
    public void hSet(String key, String subkey, Object value) {
        synchronized (objectCache) {
            Map<String, Object> submap = (Map<String, Object>) (key, k -> new ConcurrentHashMap<>());
            (subkey, value);
        }
    }

    /**
      * Get hash table data
      *
      * @param key hash table key
      * @param subkey hash table subkey
      * @return hash table value
      */
    public <T> T hGet(String key, String subkey) {
        synchronized (objectCache) {
            Map<String, Object> submap = (Map<String, Object>) (key);
            return submap != null ? (T) (subkey) : null;
        }
    }

    /**
      * Determine whether the hash table subkey exists
      *
      * @param key hash table key
      * @param subkey hash table subkey
      * @return Does it exist
      */
    public boolean hExists(String key, String subkey) {
        synchronized (objectCache) {
            Map<String, Object> submap = (Map<String, Object>) (key);
            return submap != null && (subkey);
        }
    }

    /**
      * Push data into the list
      *
      * @param key List key
      * @param value List value
      */
    public void lPush(String key, Object value) {
        synchronized (objectCache) {
            LinkedList<Object> queue = (LinkedList<Object>) (key, k -> new LinkedList<>());
            (value);
        }
    }

    /**
      * Pop up data from the list
      *
      * @param key List key
      * @return List value
      */
    public <T> T lPop(String key) {
        synchronized (objectCache) {
            LinkedList<Object> queue = (LinkedList<Object>) (key);
            return queue != null && !() ? (T) () : null;
        }
    }

    /**
      * Delete cached data
      *
      * @param key cache key
      */
    public void del(String key) {
        (key);
    }

    /**
      * Determine whether the cache key exists
      *
      * @param key cache key
      * @return Does it exist
      */
    public boolean exists(String key) {
        return (key);
    }
}