使用Java处理大数据避坑指南

本文主要是介绍使用Java处理大数据避坑指南，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

视频地址 https://www.bilibili.com/video/BV1MZ4y1S741

最近接到一个任务：对一大批数据进行处理，先使用接口把数据拉取到本地，然后再写脚本进行处理，数据量大概有几百万，所以单线程就不满足了。

处理逻辑其实不重要，毕竟这个没有重用性，就单纯来说说这个多线程是如何来处理数据的，以及我在里面所踩的坑。

需要请求一个接口去拿数据，接口的参数有这三个page 、 pageSize、 projectId，你可以理解这个接口就是一个列表的接口

一个接口，只要请求参数对了，那么结果应该就没问题了。即便是还有问题，那其实也不重要了，我们只是调用方。

所以我把每个请求的参数存入数据库，通过判断这个请求参数，来判断我们请求的数据是否有问题。（这个是在我拉取数据失败N次想到的办法）

yxx_request表

CREATE TABLE `yxx_request` (`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'id',`page` varchar(10) COLLATE utf8mb4_unicode_ci NOT NULL,`page_size` varchar(10) COLLATE utf8mb4_unicode_ci NOT NULL,`project_id` varchar(50) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '项目ID',`threadName` varchar(50) COLLATE utf8mb4_unicode_ci DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=50400 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='111';

下面就用代码模拟有50个不同的projectId，每个projectId下有50页数据。

一、单线程

伪代码如下：

public void xdxTestOne() {System.out.println("开始了......");final int pageCount = 50;List<String> projectIds = new ArrayList<>();for (int i = 0;i < 50; i++) {projectIds.add(UUID.randomUUID().toString());}HashMap params = new HashMap();params.put("pageSize", 100);for (String projectId : projectIds) {params.put("projectId", projectId);for (int i = 1; i <= pageCount; i++) {params.put("page", i);System.out.println(JSON.toJSONString(params));params.put("threadName", Thread.currentThread().getName());yxxBatchDataMapper.insertRequest(params);// TODO 通过接口去获取数据，然后存储到本地}}System.out.println("结束了......");
}

insertRequest

INSERT INTO yxx_request (page, page_size, project_id, threadName) 
VALUES (#{page}, #{pageSize}, #{projectId}, #{threadName})

上面的代码很简单，数据也没有任何问题，就不做展示了。

二、多线程

把上面的代码改造成多线程如下

public void xdxTestTwo() {System.out.println("开始了2......");final int pageCount = 50;List<String> projectIds = new ArrayList<>();for (int i = 0;i < 50; i++) {projectIds.add(UUID.randomUUID().toString());}HashMap params = new HashMap();params.put("pageSize", 100);ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(15, 15, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>());for (String projectId : projectIds) {params.put("projectId", projectId);threadPoolExecutor.execute(() -> {for (int i = 1; i <= pageCount; i++) {params.put("page", i);System.out.println(JSON.toJSONString(params));params.put("threadName", Thread.currentThread().getName());yxxBatchDataMapper.insertRequest(params);}});}System.out.println("结束了2......");
}

ThreadPoolExecutor 是线程池，不熟悉的小伙伴可以先看看这个 Java线程池详解

大家可以想象一下上面代码执行的结果，判断一下这样的改造是否有问题呢？

当然，既然我如此的问，那一定是有问题的，首先我们看看数据量，通过上面的描述我们是要模拟 50*50 的请求，那应该是有 2500 条请求才对的。

在这里插入图片描述
首先这个数量就对不上了，正确的次数应该 2500

在这里插入图片描述
page ， page_size，project_id，这三个字段进行分组应该是唯一的，但现在却找到了多条数据

在这里插入图片描述

明显数据出了问题，这个问题，我排查了很久，其实这是多线程一个常见的问题：多线程共享一个变量，A线程改变了数据，但B线程还在使用，解决的办法也很简单：把共享变量改成多例的

改造后的代码

public void xdxTestTwo() {System.out.println("开始了2......");final int pageCount = 50;List<String> projectIds = new ArrayList<>();for (int i = 0;i < 50; i++) {projectIds.add(UUID.randomUUID().toString());}ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(15, 15, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>());for (String projectId : projectIds) {threadPoolExecutor.execute(() -> {HashMap params = new HashMap();params.put("pageSize", 100);params.put("projectId", projectId);for (int i = 1; i <= pageCount; i++) {params.put("page", i);System.out.println(JSON.toJSONString(params));params.put("threadName", Thread.currentThread().getName());yxxBatchDataMapper.insertRequest(params);}});}System.out.println("结束了2......");
}