Java高并发实战:利用线程池和Redis实现高效数据入库

本文涉及的产品
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Tair(兼容Redis),内存型 2GB
简介: Java高并发实战:利用线程池和Redis实现高效数据入库

Java高并发实战:利用线程池和Redis实现高效数据入库

在高并发环境下进行数据入库是一项具有挑战性的任务。为了保证系统的性能和稳定性,可以利用线程池和Redis来实现数据的实时缓存和批量入库处理。本文将介绍一个具体实现,该实现能够根据设定的超时时间和最大批次处理数据入库。

主要思路

  • 实时数据缓存:接收到的数据首先存入Redis,保证数据的实时性。
  • 批量数据入库:当达到设定的超时时间或最大批次数量时,批量将数据从Redis中取出并入库。


主要组件

  • BatchDataStorageService:核心服务类,负责数据的缓存和批量入库。
  • CacheService:缓存服务类,使用Java的ConcurrentHashMap实现简易缓存。
  • RedisUtils:Redis工具类,用于数据的缓存。
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;

/**
 * <pre>
 *   数据批量入库服务
 * </pre>
 * Created by RuiXing Hou on 2021-08-05.
 *
 * @since 1.0
 */
@Component
@Slf4j
public class BatchDataStorageService implements InitializingBean
{
  /**
   * 最大批次数量
   */
  @Value("${app.db.maxBatchCount:800}")
    private int maxBatchCount;

  /**
   * 最大线程数
   */
    @Value("${app.db.maxBatchThreads:100}")
    private int maxBatchThreads;

  /**
   * 超时时间
   */
  @Value("${app.db.batchTimeout:3000}")
    private int batchTimeout;

  /**
   * 批次数量
   */
    private int batchCount = 0;

  /**
   * 批次号
   */
  private static long batchNo = 0;

  /**
  * 获取当前机器的核数
  */
  public static final int cpuNum = Runtime.getRuntime().availableProcessors();

  /**
   * 线程池定义接口
   */
    private ExecutorService executorService = null;

  /**
   * 服务器缓存工具类,下面提供源码
   */
  @Resource
  private CacheService cacheService;

  /**
   * 业务接口
   */
  @Resource
  private DeviceRealTimeService deviceRealTimeService;

  /**
   * redis工具类
   */
  @Resource
  private RedisUtils redisUtils;

  @Override
  public void afterPropertiesSet() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    // 核心线程大小
        taskExecutor.setCorePoolSize(cpuNum);
        // 最大线程大小
        taskExecutor.setMaxPoolSize(cpuNum * 2);
        // 队列最大容量
        taskExecutor.setQueueCapacity(500);
        // 当提交的任务个数大于QueueCapacity,就需要设置该参数,但spring提供的都不太满足业务场景,可以自定义一个,也可以注意不要超过QueueCapacity即可
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(60);
        taskExecutor.setThreadFactory(r -> {
            Thread thread = new Thread(r);
            if (r instanceof BatchWorker) {
                thread.setName("batch-worker-" + ((BatchWorker) r).batchKey);
            });
        taskExecutor.initialize();
        executorService = taskExecutor.getThreadPoolExecutor();
  }

  /**
   * 需要做高并发处理的类只需要调用该方法 (我用的是rabbitMq)
   *
   * @param deviceRealTimeDTO
   */
  public void saveRealTimeData(DeviceRealTimeDTO deviceRealTimeDTO) {
    final String failedCacheKey = "device:real_time:failed_records";

    try {

      String durationKey = "device:real_time:batchDuration" + batchNo;
      String batchKey = "device:real_time:batch" + batchNo;

      if (!cacheService.exists(durationKey)) {
        cacheService.put(durationKey, System.currentTimeMillis());
        new BatchTimeoutCommitThread(batchKey, durationKey, failedCacheKey).start();
      }

      cacheService.lPush(batchKey, deviceRealTimeDTO);
      if (++batchCount >= maxBatchCount) {
        // 达到最大批次,执行入库逻辑
        dataStorage(durationKey, batchKey, failedCacheKey);
      }

    } catch (Exception ex) {
      log.warn("[DB:FAILED] 设备上报记录入批处理集合异常: " + ex.getMessage() + ", DeviceRealTimeDTO: " + JSON.toJSONString(deviceRealTimeDTO), ex);
      cacheService.lPush(failedCacheKey, deviceRealTimeDTO);
    } finally {
      updateRealTimeData(deviceRealTimeDTO);
    }
  }

  /**
   * 更新实时数据
   * @param deviceRealTimeDTO 业务POJO
   */
  private void updateRealTimeData(DeviceRealTimeDTO deviceRealTimeDTO) {
    redisUtils.set("real_time:"+deviceRealTimeDTO.getDeviceId(), JSONArray.toJSONString(deviceRealTimeDTO));
  }

  /**
   *
   * @param durationKey     持续时间标识
   * @param batchKey      批次标识
   * @param failedCacheKey  错误标识
   */
  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 {
        DeviceRealTimeDTO deviceRealTimeDTO = cacheService.lPop(batchKey);
        while(deviceRealTimeDTO != null) {
          deviceRealTimeDTOList.add(deviceRealTimeDTO);
          deviceRealTimeDTO = cacheService.lPop(batchKey);
        }

        long timeMillis = System.currentTimeMillis();

        try {
          List<DeviceRealTimeEntity> deviceRealTimeEntityList = ConvertUtils.sourceToTarget(deviceRealTimeDTOList, DeviceRealTimeEntity.class);
          deviceRealTimeService.insertBatch(deviceRealTimeEntityList);
        } finally {
          cacheService.del(batchKey);
          log.info("[DB:BATCH_WORKER] 批次:" + batchKey + ",保存设备上报记录数:" + deviceRealTimeDTOList.size() + ", 耗时:" + (System.currentTimeMillis() - timeMillis) + "ms");
        }
      } catch (Exception e) {
        log.warn("[DB:FAILED] 设备上报记录批量入库失败:" + e.getMessage() + ", DeviceRealTimeDTO: " + deviceRealTimeDTOList.size(), e);
        for (DeviceRealTimeDTO deviceRealTimeDTO : deviceRealTimeDTOList) {
          cacheService.lPush(failedCacheKey, deviceRealTimeDTO);
        }
      }
    }
    }

  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);
    }

    public void run() {
      try {
        Thread.sleep(batchTimeout);
      } catch (InterruptedException e) {
        log.error("[DB] 内部错误,直接提交:" + e.getMessage());
      }

      if (cacheService.exists(durationKey)) {
        // 达到最大批次的超时间,执行入库逻辑
        dataStorage(durationKey, batchKey, failedCacheKey);
      }
    }

  }

}

package io.jack.service;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

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

@Component
@Scope("singleton")
public class CacheService implements InitializingBean {

    private Map<String, Object> objectCache = new ConcurrentHashMap<>();

    private Map<String, AtomicLong> statCache = new ConcurrentHashMap<>();

    @Override
    public void afterPropertiesSet() {
        statCache.put("terminals", new AtomicLong(0));
        statCache.put("connections", new AtomicLong(0));
    }

    public long incr(String statName) {
        if (!statCache.containsKey(statName))
            statCache.put(statName, new AtomicLong(0));
        return statCache.get(statName).incrementAndGet();
    }

    public long decr(String statName) {
        if (!statCache.containsKey(statName))
            statCache.put(statName, new AtomicLong(0));
        return statCache.get(statName).decrementAndGet();
    }

    public long stat(String statName) {
        if (!statCache.containsKey(statName))
            statCache.put(statName, new AtomicLong(0));
        return statCache.get(statName).get();
    }

    public <T> void put(String key, T object) {
        objectCache.put(key, object);
    }

    public <T> T get(String key) {
        return (T) objectCache.get(key);
    }

    public void remove(String key) {
        objectCache.remove(key);
    }

    public void hSet(String key, String subkey, Object value) {
        synchronized (objectCache) {
            HashMap<String, Object> submap = (HashMap<String, Object>) objectCache.get(key);
            if (submap == null) {
                submap = new HashMap<>();
                objectCache.put(key, submap);
            }
            submap.put(subkey, value);
        }
    }

    public <T> T hGet(String key, String subkey) {
        synchronized (objectCache) {
            HashMap<String, Object> submap = (HashMap<String, Object>) objectCache.get(key);
            if (submap != null) {
                return (T) submap.get(subkey);
            }
            return null;
        }
    }

    public boolean hExists(String key, String subkey) {
        synchronized (objectCache) {
            HashMap<String, Object> submap = (HashMap<String, Object>) objectCache.get(key);
            if (submap != null) {
                return submap.containsKey(subkey);
            }
            return false;
        }
    }

    public void lPush(String key, Object value) {
        synchronized (objectCache) {
            LinkedList queue = (LinkedList) objectCache.get (key);
            if (queue == null) {
                queue = new LinkedList();
                objectCache.put(key, queue);
            }
            queue.addLast(value);
        }
    }

    public <T> T lPop(String key) {
        synchronized (objectCache) {
            LinkedList queue = (LinkedList) objectCache.get (key);
            if (queue != null) {
                if (!queue.isEmpty()) {
                    return (T)queue.removeLast();
                }
                objectCache.remove(key);
            }
            return null;
        }
    }

    public void del(String key) {
        objectCache.remove(key);
    }

    public boolean exists(String key) {
        return objectCache.containsKey(key);
    }

    public void dump() {

    }
}

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
打赏
0
0
0
0
22
分享
相关文章
面试中的难题:线程异步执行后如何共享数据?
本文通过一个面试故事,详细讲解了Java中线程内部开启异步操作后如何安全地共享数据。介绍了异步操作的基本概念及常见实现方式(如CompletableFuture、ExecutorService),并重点探讨了volatile关键字、CountDownLatch和CompletableFuture等工具在线程间数据共享中的应用,帮助读者理解线程安全和内存可见性问题。通过这些方法,可以有效解决多线程环境下的数据共享挑战,提升编程效率和代码健壮性。
43 6
|
1月前
|
java异步判断线程池所有任务是否执行完
通过上述步骤,您可以在Java中实现异步判断线程池所有任务是否执行完毕。这种方法使用了 `CompletionService`来监控任务的完成情况,并通过一个独立线程异步检查所有任务的执行状态。这种设计不仅简洁高效,还能确保在大量任务处理时程序的稳定性和可维护性。希望本文能为您的开发工作提供实用的指导和帮助。
109 17
面试必问的多线程优化技巧与实战
多线程编程是现代软件开发中不可或缺的一部分,特别是在处理高并发场景和优化程序性能时。作为Java开发者,掌握多线程优化技巧不仅能够提升程序的执行效率,还能在面试中脱颖而出。本文将从多线程基础、线程与进程的区别、多线程的优势出发,深入探讨如何避免死锁与竞态条件、线程间的通信机制、线程池的使用优势、线程优化算法与数据结构的选择,以及硬件加速技术。通过多个Java示例,我们将揭示这些技术的底层原理与实现方法。
122 3
高并发场景秒杀抢购超卖Bug实战重现
在电商平台的秒杀活动中,高并发场景下的抢购超卖Bug是一个常见且棘手的问题。一旦处理不当,不仅会引发用户投诉,还会对商家的信誉和利益造成严重损害。本文将详细介绍秒杀抢购超卖Bug的背景历史、业务场景、底层原理以及Java代码实现,旨在帮助开发者更好地理解和解决这一问题。
105 12
深入理解Java中的线程池实现原理及其性能优化####
本文旨在揭示Java中线程池的核心工作机制,通过剖析其背后的设计思想与实现细节,为读者提供一份详尽的线程池性能优化指南。不同于传统的技术教程,本文将采用一种互动式探索的方式,带领大家从理论到实践,逐步揭开线程池高效管理线程资源的奥秘。无论你是Java并发编程的初学者,还是寻求性能调优技巧的资深开发者,都能在本文中找到有价值的内容。 ####
Java中的线程池优化实践####
本文深入探讨了Java中线程池的工作原理,分析了常见的线程池类型及其适用场景,并通过实际案例展示了如何根据应用需求进行线程池的优化配置。文章首先介绍了线程池的基本概念和核心参数,随后详细阐述了几种常见的线程池实现(如FixedThreadPool、CachedThreadPool、ScheduledThreadPool等)的特点及使用场景。接着,通过一个电商系统订单处理的实际案例,分析了线程池参数设置不当导致的性能问题,并提出了相应的优化策略。最终,总结了线程池优化的最佳实践,旨在帮助开发者更好地利用Java线程池提升应用性能和稳定性。 ####
Java中的线程池深度解析####
本文深入探讨了Java并发编程中的核心组件——线程池,从其基本概念、工作原理、核心参数解析到应用场景与最佳实践,全方位剖析了线程池在提升应用性能、资源管理和任务调度方面的重要作用。通过实例演示和性能对比,揭示合理配置线程池对于构建高效Java应用的关键意义。 ####
线程池关闭时未完成的任务如何保证数据的一致性?
保证线程池关闭时未完成任务的数据一致性需要综合运用多种方法和机制。通过备份与恢复、事务管理、任务状态记录与恢复、数据同步与协调、错误处理与补偿、监控与预警等手段的结合,以及结合具体业务场景进行分析和制定策略,能够最大程度地确保数据的一致性,保障系统的稳定运行和业务的顺利开展。同时,不断地优化和改进这些方法和机制,也是提高系统性能和可靠性的重要途径。
143 62
Java多线程编程中的并发容器:深入解析与实战应用####
在本文中,我们将探讨Java多线程编程中的一个核心话题——并发容器。不同于传统单一线程环境下的数据结构,并发容器专为多线程场景设计,确保数据访问的线程安全性和高效性。我们将从基础概念出发,逐步深入到`java.util.concurrent`包下的核心并发容器实现,如`ConcurrentHashMap`、`CopyOnWriteArrayList`以及`BlockingQueue`等,通过实例代码演示其使用方法,并分析它们背后的设计原理与适用场景。无论你是Java并发编程的初学者还是希望深化理解的开发者,本文都将为你提供有价值的见解与实践指导。 --- ####
Java 多线程并发控制:深入理解与实战应用
《Java多线程并发控制:深入理解与实战应用》一书详细解析了Java多线程编程的核心概念、并发控制技术及其实战技巧,适合Java开发者深入学习和实践参考。
95 8

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等