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

2024-04-16 15:32

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

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


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

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


需要请求一个接口去拿数据,接口的参数有这三个pagepageSizeprojectId,你可以理解这个接口就是一个列表的接口

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

所以我把每个请求的参数存入数据库,通过判断这个请求参数,来判断我们请求的数据是否有问题。(这个是在我拉取数据失败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......");
}

三、问题

其实说白了这个问题的解决办法是很简单的,只是多线程共用了单例变量,但在写这些个代码还是遇到了很多的问题


3-1、如何启动运行

上面代码我们是写好了,怎么运行它呢?因为我们现在都是基于SpringBoot开发,我想大家应该也会和我一样,直接写一个controller,然后一个http请求就完事了。

这看似没有任何问题,但实际存在大问题,我最开始就是这样做的,然后数据库一直存在重复的数据(共享变量也是导致数据重复的原因之一),后面我在方法的第一行打印了一个日志,发现这个方法被请求了多次

使用postman和浏览器都是相同的结果,应该是http的重试机制导致的。

解决的办法也有两个

  • 一个是使用定时任务去执行
  • 一个是方法异步,调用成功直接返回结果 (推荐)

3-2、多线程共享变量

这个上面已经解释了


3-3、数据分页

字面意思,就是sql分页 LIMIT page, pageSize,但现在我们写代码都早已习惯了各种分页插件,没有自己去写过,下面就是我传递的page, pageSize

  • 1 500
  • 2 500
  • 3 500

正确的传参(还是在请教老大看出来的)

  • 0 500
  • 501 500
  • 1001 500

3-4、提升速度

  • 一般处理数据,可能会单独的去创建一些零时表,然后对这个表去进行增删改查操作,这时候我们经常会忘记给这个零时表加索引

  • 操作数据库能批量就批量,当然要注意数据量 500-1000条/次 还是ok的

  • 能在本地就在本地跑,一般我们自己的电脑都是16G+,服务器除非是线上,测试环境都很小,我在测试环境开50个线程就挂机了,在本地就不会。


3-N、其它

这里给处理数据的一些建议

  • 处理线上数据一定、一定、一定要备份
  • 先写出单线程,测试执行没问题了再改多线程(熟悉之后改起来很快)
  • 在处理数据的时候不知道会出现什么问题,最好给数据加一个标识,这样处理中断不至于从头开始

关注我吧,一起学习、一起成长

在这里插入图片描述

这篇关于使用Java处理大数据避坑指南的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/909184

相关文章

springboot集成easypoi导出word换行处理过程

《springboot集成easypoi导出word换行处理过程》SpringBoot集成Easypoi导出Word时,换行符n失效显示为空格,解决方法包括生成段落或替换模板中n为回车,同时需确... 目录项目场景问题描述解决方案第一种:生成段落的方式第二种:替换模板的情况,换行符替换成回车总结项目场景s

SpringBoot集成redisson实现延时队列教程

《SpringBoot集成redisson实现延时队列教程》文章介绍了使用Redisson实现延迟队列的完整步骤,包括依赖导入、Redis配置、工具类封装、业务枚举定义、执行器实现、Bean创建、消费... 目录1、先给项目导入Redisson依赖2、配置redis3、创建 RedissonConfig 配

SpringBoot中@Value注入静态变量方式

《SpringBoot中@Value注入静态变量方式》SpringBoot中静态变量无法直接用@Value注入,需通过setter方法,@Value(${})从属性文件获取值,@Value(#{})用... 目录项目场景解决方案注解说明1、@Value("${}")使用示例2、@Value("#{}"php

SpringBoot分段处理List集合多线程批量插入数据方式

《SpringBoot分段处理List集合多线程批量插入数据方式》文章介绍如何处理大数据量List批量插入数据库的优化方案:通过拆分List并分配独立线程处理,结合Spring线程池与异步方法提升效率... 目录项目场景解决方案1.实体类2.Mapper3.spring容器注入线程池bejsan对象4.创建

线上Java OOM问题定位与解决方案超详细解析

《线上JavaOOM问题定位与解决方案超详细解析》OOM是JVM抛出的错误,表示内存分配失败,:本文主要介绍线上JavaOOM问题定位与解决方案的相关资料,文中通过代码介绍的非常详细,需要的朋... 目录一、OOM问题核心认知1.1 OOM定义与技术定位1.2 OOM常见类型及技术特征二、OOM问题定位工具

PHP轻松处理千万行数据的方法详解

《PHP轻松处理千万行数据的方法详解》说到处理大数据集,PHP通常不是第一个想到的语言,但如果你曾经需要处理数百万行数据而不让服务器崩溃或内存耗尽,你就会知道PHP用对了工具有多强大,下面小编就... 目录问题的本质php 中的数据流处理:为什么必不可少生成器:内存高效的迭代方式流量控制:避免系统过载一次性

Python正则表达式匹配和替换的操作指南

《Python正则表达式匹配和替换的操作指南》正则表达式是处理文本的强大工具,Python通过re模块提供了完整的正则表达式功能,本文将通过代码示例详细介绍Python中的正则匹配和替换操作,需要的朋... 目录基础语法导入re模块基本元字符常用匹配方法1. re.match() - 从字符串开头匹配2.

基于 Cursor 开发 Spring Boot 项目详细攻略

《基于Cursor开发SpringBoot项目详细攻略》Cursor是集成GPT4、Claude3.5等LLM的VSCode类AI编程工具,支持SpringBoot项目开发全流程,涵盖环境配... 目录cursor是什么?基于 Cursor 开发 Spring Boot 项目完整指南1. 环境准备2. 创建

Python使用FastAPI实现大文件分片上传与断点续传功能

《Python使用FastAPI实现大文件分片上传与断点续传功能》大文件直传常遇到超时、网络抖动失败、失败后只能重传的问题,分片上传+断点续传可以把大文件拆成若干小块逐个上传,并在中断后从已完成分片继... 目录一、接口设计二、服务端实现(FastAPI)2.1 运行环境2.2 目录结构建议2.3 serv

C#实现千万数据秒级导入的代码

《C#实现千万数据秒级导入的代码》在实际开发中excel导入很常见,现代社会中很容易遇到大数据处理业务,所以本文我就给大家分享一下千万数据秒级导入怎么实现,文中有详细的代码示例供大家参考,需要的朋友可... 目录前言一、数据存储二、处理逻辑优化前代码处理逻辑优化后的代码总结前言在实际开发中excel导入很