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

本文涉及的产品
云原生内存数据库 Tair,内存型 2GB
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Redis 版,经济版 1GB 1个月
简介: 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
目录
相关文章
|
6天前
|
Java 调度
Java线程的六种状态
Java线程有六种状态: 初始(NEW)、运行(RUNNABLE)、阻塞(BLOCKED)、等待(WAITING)、超时等待(TIMED_WAITING)、终止(TERMINATED)。
20 1
|
1天前
|
Java 关系型数据库 数据库
实时计算 Flink版操作报错合集之拉取全量数据时,如何解决Checkpoint失败并且报错为 "java.lang.OutOfMemoryError: Java heap space"
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
3天前
|
安全 Java 开发者
Java并发编程中的线程安全性与性能优化
在Java编程中,处理并发问题是至关重要的。本文探讨了Java中线程安全性的概念及其在性能优化中的重要性。通过深入分析多线程环境下的共享资源访问问题,结合常见的并发控制手段和性能优化技巧,帮助开发者更好地理解和应对Java程序中的并发挑战。 【7月更文挑战第14天】
|
3天前
|
监控 Java API
Java并发编程之线程池深度解析
【7月更文挑战第14天】在Java并发编程领域,线程池是提升性能、管理资源的关键工具。本文将深入探讨线程池的核心概念、内部工作原理以及如何有效使用线程池来处理并发任务,旨在为读者提供一套完整的线程池使用和优化策略。
|
6天前
|
缓存 安全 Java
Java中线程池如何管理?
【7月更文挑战第11天】Java中线程池如何管理?
15 2
|
6天前
|
存储 安全 算法
深入理解Java并发编程:线程安全与性能优化
【5月更文挑战第72天】 在现代软件开发中,尤其是Java应用开发领域,并发编程是一个无法回避的重要话题。随着多核处理器的普及,合理利用并发机制对于提高软件性能、响应速度和资源利用率具有重要意义。本文旨在探讨Java并发编程的核心概念、线程安全的策略以及性能优化技巧,帮助开发者构建高效且可靠的并发应用。通过实例分析和理论阐述,我们将揭示在高并发环境下如何平衡线程安全与系统性能之间的关系,并提出一系列最佳实践方法。
|
5天前
|
Java 调度
java中线程的6种状态
java中线程的6种状态
|
5天前
|
算法 Java 开发者
Java中的多线程编程技巧与实践
在现代软件开发中,多线程编程成为提升应用程序性能和响应能力的关键技术之一。本文将深入探讨Java语言中多线程编程的基础概念、常见问题及其解决方案,帮助开发者更好地理解和应用多线程技术。 【7月更文挑战第12天】
7 0
|
2月前
|
消息中间件 Java Linux
2024年最全BATJ真题突击:Java基础+JVM+分布式高并发+网络编程+Linux(1),2024年最新意外的惊喜
2024年最全BATJ真题突击:Java基础+JVM+分布式高并发+网络编程+Linux(1),2024年最新意外的惊喜
|
17天前
|
存储 NoSQL Java
探索Java分布式锁:在高并发环境下的同步访问实现与优化
【6月更文挑战第30天】Java分布式锁在高并发下确保数据一致性,通过Redis的SETNX、ZooKeeper的临时节点、数据库操作等方式实现。优化策略包括锁超时重试、续期、公平性及性能提升,关键在于平衡同步与效率,适应大规模分布式系统的需求。
29 1