• Java实现从Redis中批量读取数据

    一.背景概述

    本周接到一个新的需求：从用户dau日志文件中读取用户uid，然后到Redis中获取对应的用户数据。用户的uid存储于login_day_20220913.txt文件，共1亿2千多万条数据，数量达1.4G。

    要求：尽量在2小时内获得结果，在数据处理过程中，Redis服务器QPS尽量低，不超过某个阈值，不然会触发监控报警。数据从Redis从库读取，只提供一个端口。

    二.分析与实现

    由于之前做过相同数据量的统计需求，所以从一开始就确定单线程完成此次数据处理也是可以的。实际上，对多线程和并发的使用需要慎之又慎，特别是在业务繁忙的系统或环境下。

    接触Redis的朋友都知道，Redis是支持批量读取的,其中常用的两个方法：mget()和hmget()。

    本次处理的数据不是哈希结构，所以确定使用mget()。

    此时，我自然而然地问了同事一个问题，那就是mget批量处理数据的最佳参数范围是多少？因为mget()接受一个字符串数组参数，也就是说字符串数组的长度最佳为多少？

    同事并没有给我明确的答案，只是说他们日常每批次处理10000条，建议我自己可以尝试一下，于是我打算试试50000条数据。

    主要代码如下：

    1. package com.sina.weibo;
    2.  
    3. import com.sina.weibo.util.FileUtils;
    4. import com.sina.weibo.util.ListUtil;
    5. import org.apache.commons.lang3.time.StopWatch;
    6. import redis.clients.jedis.Jedis;
    7.  
    8. import java.util.ArrayList;
    9. import java.util.LinkedHashSet;
    10. import java.util.List;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. public class Application {
    14.     /** dau数据读取路径 */
    15.     private static String dauDataPath = "/data1/sinawap/var/logs/wapcommon/place/user_position/dau/login_day_20220913.txt";
    16.  
    17.     /** 结果输出路径 */
    18.     private static String outputPath = "/data1/bingqing5/importcampusdata/output/campus_data.txt";
    19.  
    20.     /** 已处理过的uid数据存储路径 */
    21.     private static String processedUidDataPath = "/data1/bingqing5/importcampusdata/process/processed_uid.txt";
    22.  
    23.     public static void main(String[] args) {
    24.         StopWatch stopWatch = new StopWatch();
    25.         // 开始时间
    26.         stopWatch.start();
    27.         System.out.println("================程序开始===============");
    28.         transfer(dauDataPath, processedUidDataPath, outputPath);
    29.         System.out.println("================程序结束===============");
    30.         // 结束时间
    31.         stopWatch.stop();
    32.         // 统计执行时间（秒）
    33.         System.out.println("执行时长：" + stopWatch.getTime(TimeUnit.SECONDS) + " 秒.");
    34.     }
    35.  
    36.     private static void transfer(String dauDataPath, String processedUidDataPath, String outputPath) {
    37.         List dauDataList = FileUtils.readInfoFromFile(dauDataPath);
    38.         List> bucket = ListUtil.splitList(dauDataList, 50000);
    39.         Jedis jedis = new Jedis("rdsxxxxx.xxxx.xxxx.xxxx.com.cn",50000);
    40.         List processedUidDataList = FileUtils.readInfoFromFile(processedUidDataPath);
    41.         LinkedHashSet linkedHashSet = ListUtil.getLinkedHashSet(processedUidDataList);
    42.         for (List list : bucket) {
    43.             List jsonStrList = jedis.mget(list.toArray(new String[list.size()]));
    44.             for (int i = 0; i < list.size(); i++) {
    45.                 if (!linkedHashSet.contains(list.get(i))) {
    46.                     String uid = list.get(i);
    47.                     FileUtils.appendInfoToFile(processedUidDataPath, uid);
    48.                     String jsonStr = jsonStrList.get(i);
    49.                     if (jsonStr == null || jsonStr == "") continue;
    50.                     String content = uid + "\t" + jsonStr;
    51.                     FileUtils.appendInfoToFile(outputPath, content);
    52.                 }
    53.             }
    54.             System.out.println(list.size());
    55.         }
    56.     }
    57. }

    三.发现问题与屡次改进

    3.1.QPS过高而且波动很大

    上述代码上线后没多久，就被同事找来，说QPS过高，开始的时候瞬间达到近100k，之后稳定在70k~100k之间。因为担心影响其他业务，于是把jar包暂停，着手优化。

    于是，我多次修改如下代码：

    List> bucket = ListUtil.splitList(dauDataList, 50000);

    将50000，调整为10000,5000,1000,500,100等值逐一尝试。

    QPS确实逐步降下来了，但是即便是每次处理1000条，QPS也有40K左右。

    3.2.程序中断，抛异常

    最终以每批次读取500条数据，将代码上线。但是程序总是中断报错，抛出异常：

    而这时候已处理的数据量达到几千万条。

    最初怀疑是因为jedis对象没有调用close方法，于是修改代码如下：

    1. package com.sina.weibo;
    2.  
    3. import com.sina.weibo.util.FileUtils;
    4. import com.sina.weibo.util.ListUtil;
    5. import org.apache.commons.lang3.time.StopWatch;
    6. import redis.clients.jedis.Jedis;
    7.  
    8. import java.util.ArrayList;
    9. import java.util.LinkedHashSet;
    10. import java.util.List;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. public class Application {
    14.     /** dau数据读取路径 */
    15.     private static String dauDataPath = "/data1/sinawap/var/logs/wapcommon/place/user_position/dau/login_day_20220913.txt";
    16.  
    17.     /** 结果输出路径 */
    18.     private static String outputPath = "/data1/bingqing5/importcampusdata/output/campus_data.txt";
    19.  
    20.     /** 已处理过的uid数据存储路径 */
    21.     private static String processedUidDataPath = "/data1/bingqing5/importcampusdata/process/processed_uid.txt";
    22.  
    23.     public static void main(String[] args) {
    24.         StopWatch stopWatch = new StopWatch();
    25.         // 开始时间
    26.         stopWatch.start();
    27.         System.out.println("================程序开始===============");
    28.         transfer(dauDataPath, processedUidDataPath, outputPath);
    29.         System.out.println("================程序结束===============");
    30.         // 结束时间
    31.         stopWatch.stop();
    32.         // 统计执行时间（秒）
    33.         System.out.println("执行时长：" + stopWatch.getTime(TimeUnit.SECONDS) + " 秒.");
    34.     }
    35.  
    36.     private static void transfer(String dauDataPath, String processedUidDataPath, String outputPath) {
    37.         List dauDataList = FileUtils.readInfoFromFile(dauDataPath);
    38.         List> bucket = ListUtil.splitList(dauDataList, 50000);
    39.         List processedUidDataList = FileUtils.readInfoFromFile(processedUidDataPath);
    40.         LinkedHashSet linkedHashSet = ListUtil.getLinkedHashSet(processedUidDataList);
    41.         for (List list : bucket) {
    42.             Jedis jedis = new Jedis(rdsxxxxx.xxxx.xxxx.xxxx.com.cn", 50000);
    43.             List jsonStrList = jedis.mget(list.toArray(new String[list.size()]));
    44.             for (int i = 0; i < list.size(); i++) {
    45.                 if (!linkedHashSet.contains(list.get(i))) {
    46.                     String uid = list.get(i);
    47.                     FileUtils.appendInfoToFile(processedUidDataPath, uid);
    48.                     String jsonStr = jsonStrList.get(i);
    49.                     if (jsonStr == null || jsonStr == "") continue;
    50.                     String content = uid + "\t" + jsonStr;
    51.                     FileUtils.appendInfoToFile(outputPath, content);
    52.                 }
    53.             }
    54.             jedis.close();
    55.             System.out.println(list.size());
    56.         }
    57.     }
    58. }

     修改后跑程序依旧没有任何改善，继续修改，代码如下：

    1. package com.sina.weibo;
    2.  
    3. import com.sina.weibo.util.FileUtils;
    4. import com.sina.weibo.util.ListUtil;
    5. import org.apache.commons.lang3.time.StopWatch;
    6. import redis.clients.jedis.Jedis;
    7.  
    8. import java.util.ArrayList;
    9. import java.util.LinkedHashSet;
    10. import java.util.List;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. public class A {
    14.     /** dau数据读取路径 */
    15.     private static String dauDataPath = "/data1/sinawap/var/logs/wapcommon/place/user_position/dau/login_day_20220913.txt";
    16.  
    17.     /** 结果输出路径 */
    18.     private static String outputPath = "/data1/bingqing5/importcampusdata/output/campus_data.txt";
    19.  
    20.     /** 已处理过的uid数据存储路径 */
    21.     private static String processedUidDataPath = "/data1/bingqing5/importcampusdata/process/processed_uid.txt";
    22.  
    23.     public static void main(String[] args) {
    24.         StopWatch stopWatch = new StopWatch();
    25.         // 开始时间
    26.         stopWatch.start();
    27.         System.out.println("================程序开始===============");
    28.         transfer(dauDataPath, processedUidDataPath, outputPath);
    29.         System.out.println("================程序结束===============");
    30.         // 结束时间
    31.         stopWatch.stop();
    32.         // 统计执行时间（秒）
    33.         System.out.println("执行时长：" + stopWatch.getTime(TimeUnit.SECONDS) + " 秒.");
    34.     }
    35.  
    36.     private static void transfer(String dauDataPath, String processedUidDataPath, String outputPath) {
    37.         List dauDataList = FileUtils.readInfoFromFile(dauDataPath);
    38.         List> bucket = ListUtil.splitList(dauDataList, 50000);
    39.         List processedUidDataList = FileUtils.readInfoFromFile(processedUidDataPath);
    40.         LinkedHashSet linkedHashSet = ListUtil.getLinkedHashSet(processedUidDataList);
    41.         for (List list : bucket) {
    42.             Jedis jedis = new Jedis("rdsxxxxx.xxxx.xxxx.xxxx.com.cn", 50000);
    43.             List jsonStrList = jedis.mget(list.toArray(new String[list.size()]));
    44.             for (int i = 0; i < list.size(); i++) {
    45.                 if (!linkedHashSet.contains(list.get(i))) {
    46.                     String uid = list.get(i);
    47.                     FileUtils.appendInfoToFile(processedUidDataPath, uid);
    48.                     String jsonStr = jsonStrList.get(i);
    49.                     if (jsonStr == null || jsonStr == "") continue;
    50.                     String content = uid + "\t" + jsonStr;
    51.                     FileUtils.appendInfoToFile(outputPath, content);
    52.                 }
    53.             }
    54.             try {
    55.                 Thread.sleep(100);
    56.             } catch (InterruptedException e) {
    57.                 e.printStackTrace();
    58.             } finally {
    59.                 jedis.close();
    60.             }
    61.             System.out.println(list.size());
    62.         }
    63.     }
    64. }

    上线以后，观测发现QPS区域稳定，但是程序会空跑，也就是从头开始将已处理的数据也要逐一读取一次，很多时候都没有跑到上次程序处理的地方就已经被迫退出。

    linkedHashSet本来是用来标记上次程序运行停止的地方，但是似乎并没有完全发挥作用。

    于是修改代码，加入一个新的list集合，用于存放还没有处理过的数据，代码如下：

    1. package com.sina.weibo;
    2.  
    3. import com.sina.weibo.util.FileUtils;
    4. import com.sina.weibo.util.ListUtil;
    5. import org.apache.commons.lang3.time.StopWatch;
    6. import redis.clients.jedis.Jedis;
    7.  
    8. import java.util.ArrayList;
    9. import java.util.LinkedHashSet;
    10. import java.util.List;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. /**
    14.  * @author bingqing5 
    15.  * @date 2022/09/14 15:00
    16.  * @version 1.0
    17.  */
    18.  
    19. public class Application {
    20.  
    21.     /** dau数据读取路径 */
    22.     private static String dauDataPath = "/data1/sinawap/var/logs/wapcommon/place/user_position/dau/login_day_20220913.txt";
    23.  
    24.     /** 结果输出路径 */
    25.     private static String outputPath = "/data1/bingqing5/importcampusdata/output/campus_data.txt";
    26.     
    27.     /** 已处理过的uid数据存储路径 */
    28.     private static String processedUidDataPath = "/data1/bingqing5/importcampusdata/process/processed_uid.txt";
    29.  
    30.     public static void main(String[] args) {
    31.         StopWatch stopWatch = new StopWatch();
    32.         // 开始时间
    33.         stopWatch.start();
    34.         System.out.println("================程序开始===============");
    35. //        transfer(dauDataPath, processedUidDataPath, outputPath);
    36.         List dauDataList = FileUtils.readInfoFromFile(dauDataPath);
    37. //        List> bucket = ListUtil.splitList(dauDataList, 50000);
    38. //        Jedis jedis = new Jedis("rdsxxxxx.xxxx.xxxx.xxxx.com.cn", 50000);
    39.         List processedUidDataList = FileUtils.readInfoFromFile(processedUidDataPath);
    40.         LinkedHashSet linkedHashSet = ListUtil.getLinkedHashSet(processedUidDataList);
    41.         List uidList = new ArrayList<>();
    42.         for (String uid : dauDataList) {
    43.             if (linkedHashSet.contains(uid)) {
    44.                 continue;
    45.             } else {
    46.                 uidList.add(uid);
    47.             }
    48.         }
    49.  
    50.         List> bucket;
    51.         if (uidList.size() != 0) {
    52.             bucket = ListUtil.splitList(uidList, 10000);
    53.         } else {
    54.             bucket = new ArrayList<>();
    55.         }
    56.  
    57.         for (List list : bucket) {
    58.             Jedis jedis = new Jedis("rdsxxxxx.xxxx.xxxx.xxxx.com.cn", 50000);
    59.             List jsonStrList = jedis.mget(list.toArray(new String[list.size()]));
    60.             for (int i = 0; i < list.size(); i++) {
    61.                 if (!linkedHashSet.contains(list.get(i))) {
    62.                     String uid = list.get(i);
    63.                     FileUtils.appendInfoToFile(processedUidDataPath, uid);
    64.                     String jsonStr = jsonStrList.get(i);
    65.                     if (jsonStr == null || jsonStr == "") continue;
    66.                     String content = uid + "\t" + jsonStr;
    67.                     FileUtils.appendInfoToFile(outputPath, content);
    68.                 }
    69.             }
    70.             try {
    71.                 Thread.sleep(100);
    72.             } catch (InterruptedException e) {
    73.                 e.printStackTrace();
    74.             } finally {
    75.                 jedis.close();
    76.             }
    77.             System.out.println(list.size());
    78.         }
    79.         System.out.println("================程序结束===============");
    80.         // 结束时间
    81.         stopWatch.stop();
    82.         // 统计执行时间（秒）
    83.         System.out.println("执行时长：" + stopWatch.getTime(TimeUnit.SECONDS) + " 秒.");
    84.     }
    85.  
    86.  
    87. }

    终于这次修改后，上线代码，代码平稳运行。

    此时查看QPS，发现10000的批读取量，QPS文档在25K以下，此前同样的数据量，QPS能达到40K。

     3.3.内存消耗过大

    在上次修改后，程序平稳运行，期间我查看了机器状态，发现我跑的jar包竟然消耗了32%左右的内存，那台机器也不过62G的总内存。虽然不缺内存资源，但是还是决定趁着程序在跑的期间，回顾一下代码。

    List> bucket = ListUtil.splitList(dauDataList, 10000);

    上面这行代码是将所有的用户uid数据按照10000的大小均等分割，每次遍历，要重复创建同一类Jedis对象，也会消耗大量内存。

    另外，下面这段程序：

    1.  List uidList = new ArrayList<>();
    2.         for (String uid : dauDataList) {
    3.             if (linkedHashSet.contains(uid)) {
    4.                 continue;
    5.             } else {
    6.                 uidList.add(uid);
    7.             }
    8.         }

    已经对处理过的数据做过筛选，在循环中再次做如下判断：

    1. if (!linkedHashSet.contains(list.get(i))) {
    2.                    
    3. }

    也是多次一举，会增加耗时。

    综合以上考虑，我做了修改，代码如下：

    1. package com.sina.weibo;
    2.  
    3. import com.sina.weibo.util.FileUtils;
    4. import com.sina.weibo.util.ListUtil;
    5. import org.apache.commons.lang3.time.StopWatch;
    6. import redis.clients.jedis.Jedis;
    7.  
    8. import java.util.ArrayList;
    9. import java.util.LinkedHashSet;
    10. import java.util.List;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. /**
    14.  * @author bingqing5
    15.  * @date 2022/09/14 15:00
    16.  * @version 1.0
    17.  */
    18.  
    19. public class Application {
    20.  
    21.     /** dau数据读取路径 */
    22.     private static String dauDataPath = "/data1/sinawap/var/logs/wapcommon/place/user_position/dau/login_day_20220913.txt";
    23.  
    24.     /** 结果输出路径 */
    25. //    private static String outputPath = "/data1/bingqing5/redis_test/output/campus_data.txt";
    26.     private static String outputPath = "/data1/bingqing/redis_test/output/campus_data.txt";
    27.  
    28.     /** 已处理过的uid数据存储路径 */
    29. //    private static String processedUidDataPath = "/data1/bingqing5/redis_test/process/processed_uid.txt";
    30.     private static String processedUidDataPath = "/data1/bingqing/redis_test/process/processed_uid.txt";
    31.  
    32.     public static void main(String[] args) {
    33.         StopWatch stopWatch = new StopWatch();
    34.         // 开始时间
    35.         stopWatch.start();
    36.         System.out.println("================程序开始===============");
    37.         transfer(dauDataPath, processedUidDataPath, outputPath);
    38.         System.out.println("================程序结束===============");
    39.         // 结束时间
    40.         stopWatch.stop();
    41.         // 统计执行时间（秒）
    42.         System.out.println("执行时长：" + stopWatch.getTime(TimeUnit.SECONDS) + " 秒.");
    43.     }
    44.  
    45.     private static void transfer(String dauDataPath, String processedUidDataPath, String outputPath) {
    46.         List dauDataList = FileUtils.readInfoFromFile(dauDataPath);
    47.         Jedis jedis = new Jedis("rdsxxxxx.xxxx.xxxx.xxxx.com.cn", 50000);
    48.         List processedUidDataList = FileUtils.readInfoFromFile(processedUidDataPath);
    49.         LinkedHashSet linkedHashSet = ListUtil.getLinkedHashSet(processedUidDataList);
    50.         List uidList = new ArrayList<>();
    51.         for (String uid : dauDataList) {
    52.             if (linkedHashSet.contains(uid)) {
    53.                 continue;
    54.             } else {
    55.                 uidList.add(uid);
    56.             }
    57.         }
    58.         List> bucket;
    59.         if (uidList.size() != 0) {
    60.             bucket = ListUtil.splitList(uidList, 50000);
    61.         } else {
    62.             bucket = new ArrayList<>();
    63.         }
    64.  
    65.         for (List list : bucket) {
    66.             List jsonStrList = jedis.mget(list.toArray(new String[list.size()]));
    67.             for (int i = 0; i < list.size(); i++) {
    68.                 String uid = list.get(i);
    69.                 FileUtils.appendInfoToFile(processedUidDataPath, uid);
    70.                 String jsonStr = jsonStrList.get(i);
    71.                 if (jsonStr == null || jsonStr == "") continue;
    72.                 String content = uid + "\t" + jsonStr;
    73.                 FileUtils.appendInfoToFile(outputPath, content);
    74.             }
    75.  
    76.             try {
    77.                 Thread.sleep(100);
    78.             } catch (InterruptedException e) {
    79.                 e.printStackTrace();
    80.             } finally {
    81.                 jedis.close();
    82.             }
    83.             System.out.println(list.size());
    84.         }
    85.     }
    86. }

    修改代码以后，替换掉原先运行的jar包，接着运行。发现内存消耗明显降低，稳定占总内存的20%。

    然后尝试修改了mget参数量，修改为50000条，再次运行程序发现QPS稳定在40K左右。

     四.总结

    本篇算是笔者刚接触Redis不久的一篇随手记。通过本次需求的开发经历，让我对Redis有了直观的了解，同时也理解了代码优化在实际生产工作和开发中的潜在价值。

    关于Redis，在快速直接从Redis读取数据的场景中，尤其是数据量大的时候，为了防止QPS过高，最好在处理一批次数据后空出一定的时间间隔，比如可以让线程暂时休眠一定时间间隔，再进行下批次读取和处理。

    关于代码优化，尽量创建可重复使用的对象，非必要不添加同类对象，避免大量创建对象带来的资源消耗，本次经历也算是很鲜明的体会到这点。

  • 相关阅读:
    IMU标定之---Allan方差
    JsonCpp 使用指导
    从零开始 Spring Boot 17：MyBatis Plus 续
    【最小的k个数】
    AQS(AbstractQueuedSynchronizer)框架之ReentrantLock
    刷代码随想录有感（66）：回溯算法——组合问题的优化（剪枝）
    Linux学习第22天：Linux中断驱动开发（一）： 突如其来
    position属性练习：画一个爱心送给你。
    【cmake实战十】c++从动态库（dll）导出类
    String的方法介绍以及实现
  • 原文地址：https://blog.csdn.net/lbq15735104044/article/details/126891550