网站首页 全球最实用的IT互联网站!

人工智能P2P分享Wind搜索发布信息网站地图标签大全

当前位置:诺佳网 > 电子/半导体 > 嵌入式技术 >

SpringBoot实现MySQL百万级数据量导出并避免OOM的解

时间:2023-03-16 13:50

人气:

作者:admin

标签:

导读:全量加载不可行,那我们的目标就是如何实现数据的分批加载了。实事上,Mysql本身支持Stream查询,我们可以通过Stream流获取数据,然后将数据逐条刷入到文件中,每次刷入文件后再从内...


前言

动态数据导出是一般项目都会涉及到的功能。它的基本实现逻辑就是从mysql查询数据,加载到内存,然后从内存创建excel或者csv,以流的形式响应给前端。

  • 参考:https://grokonez.com/spring-framework/spring-boot/excel-file-download-from-springboot-restapi-apache-poi-mysql。

SpringBoot下载excel基本都是这么干。

虽然这是个可行的方案,然而一旦mysql数据量太大,达到十万级,百万级,千万级,大规模数据加载到内存必然会引起OutofMemoryError。

要考虑如何避免OOM,一般有两个方面的思路。

一方面就是尽量不做呗,先怼产品下面几个问题啊:

  • 我们为什么要导出这么多数据呢?谁傻到去看这么大的数据啊,这个设计是不是合理的呢?
  • 怎么做好权限控制?百万级数据导出你确定不会泄露商业机密?
  • 如果要导出百万级数据,那为什么不直接找大数据或者DBA来干呢?然后以邮件形式传递不行吗?
  • 为什么要通过后端的逻辑来实现,不考虑时间成本,流量成本吗?
  • 如果通过分页导出,每次点击按钮只导2万条,分批导出难道不能满足业务需求吗?

如果产品说 “甲方是爸爸,你去和甲方说啊”,“客户说这个做出来,才考虑付尾款!”,如果客户的确缺根筋要让你这样搞, 那就只能从技术上考虑如何实现了。

从技术上讲,为了避免OOM,我们一定要注意一个原则:

不能将全量数据一次性加载到内存之中。

全量加载不可行,那我们的目标就是如何实现数据的分批加载了。实事上,Mysql本身支持Stream查询,我们可以通过Stream流获取数据,然后将数据逐条刷入到文件中,每次刷入文件后再从内存中移除这条数据,从而避免OOM。

由于采用了数据逐条刷入文件,而且数据量达到百万级,所以文件格式就不要采用excel了,excel2007最大才支持104万行的数据。这里推荐:

以csv代替excel。

考虑到当前SpringBoot持久层框架通常为JPA和mybatis,我们可以分别从这两个框架实现百万级数据导出的方案。

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

  • 项目地址:https://github.com/YunaiV/ruoyi-vue-pro
  • 视频教程:https://doc.iocoder.cn/video/

JPA实现百万级数据导出

  • 具体方案不妨参考:http://knes1.github.io/blog/2015/2015-10-19-streaming-mysql-results-using-java8-streams-and-spring-data.html。

实现项目对应:

  • https://github.com/knes1/todo

核心注解如下,需要加入到具体的Repository之上。方法的返回类型定义成Stream。Integer.MIN_VALUE告诉jdbc driver逐条返回数据。

@QueryHints(value=@QueryHint(name=HINT_FETCH_SIZE,value=""+Integer.MIN_VALUE))
@Query(value="selecttfromTodot")
StreamstreamAll();

此外还需要在Stream处理数据的方法之上添加@Transactional(readOnly = true),保证事物是只读的。

同时需要注入javax.persistence.EntityManager,通过detach从内存中移除已经使用后的对象。

@RequestMapping(value="/todos.csv",method=RequestMethod.GET)
@Transactional(readOnly=true)
publicvoidexportTodosCSV(HttpServletResponseresponse){
response.addHeader("Content-Type","application/csv");
response.addHeader("Content-Disposition","attachment;filename=todos.csv");
response.setCharacterEncoding("UTF-8");
try(StreamtodoStream=todoRepository.streamAll()){
PrintWriterout=response.getWriter();
todoStream.forEach(rethrowConsumer(todo->{
Stringline=todoToCSV(todo);
out.write(line);
out.write("
");
entityManager.detach(todo);
}));
out.flush();
}catch(IOExceptione){
log.info("Exceptionoccurred"+e.getMessage(),e);
thrownewRuntimeException("Exceptionoccurredwhileexportingresults",e);
}
}

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

  • 项目地址:https://github.com/YunaiV/yudao-cloud
  • 视频教程:https://doc.iocoder.cn/video/

MyBatis实现百万级数据导出

MyBatis实现逐条获取数据,必须要自定义ResultHandler,然后在mapper.xml文件中,对应的select语句中添加fetchSize="-2147483648"。

edd06b1c-c3b1-11ed-bfe3-dac502259ad0.png

最后将自定义的ResultHandler传给SqlSession来执行查询,并将返回的结果进行处理。

MyBatis实现百万级数据导出的具体实例

以下是基于MyBatis Stream导出的完整的工程样例,我们将通过对比Stream文件导出和传统方式导出的内存占用率的差异,来验证Stream文件导出的有效性。

我们先定义一个工具类DownloadProcessor,它内部封装一个HttpServletResponse对象,用来将对象写入到csv。

publicclassDownloadProcessor{
privatefinalHttpServletResponseresponse;

publicDownloadProcessor(HttpServletResponseresponse){
this.response=response;
StringfileName=System.currentTimeMillis()+".csv";
this.response.addHeader("Content-Type","application/csv");
this.response.addHeader("Content-Disposition","attachment;filename="+fileName);
this.response.setCharacterEncoding("UTF-8");
}

publicvoidprocessData(Erecord){
try{
response.getWriter().write(record.toString());//如果是要写入csv,需要重写toString,属性通过","分割
response.getWriter().write("
");
}catch(IOExceptione){
e.printStackTrace();
}
}
}

然后通过实现org.apache.ibatis.session.ResultHandler,自定义我们的ResultHandler,它用于获取java对象,然后传递给上面的DownloadProcessor处理类进行写文件操作:

publicclassCustomResultHandlerimplementsResultHandler{

privatefinalDownloadProcessordownloadProcessor;

publicCustomResultHandler(
DownloadProcessordownloadProcessor){
super();
this.downloadProcessor=downloadProcessor;
}

@Override
publicvoidhandleResult(ResultContextresultContext){
Authorsauthors=(Authors)resultContext.getResultObject();
downloadProcessor.processData(authors);
}
}

实体类:

publicclassAuthors{
privateIntegerid;
privateStringfirstName;

privateStringlastName;

privateStringemail;

privateDatebirthdate;

privateDateadded;

publicIntegergetId(){
returnid;
}

publicvoidsetId(Integerid){
this.id=id;
}

publicStringgetFirstName(){
returnfirstName;
}

publicvoidsetFirstName(StringfirstName){
this.firstName=firstName==null?null:firstName.trim();
}

publicStringgetLastName(){
returnlastName;
}

publicvoidsetLastName(StringlastName){
this.lastName=lastName==null?null:lastName.trim();
}

publicStringgetEmail(){
returnemail;
}

publicvoidsetEmail(Stringemail){
this.email=email==null?null:email.trim();
}

publicDategetBirthdate(){
returnbirthdate;
}

publicvoidsetBirthdate(Datebirthdate){
this.birthdate=birthdate;
}

publicDategetAdded(){
returnadded;
}

publicvoidsetAdded(Dateadded){
this.added=added;
}

@Override
publicStringtoString(){
returnthis.id+","+this.firstName+","+this.lastName+","+this.email+","+this.birthdate+","+this.added;
}
}

Mapper接口:

publicinterfaceAuthorsMapper{
ListselectByExample(AuthorsExampleexample);

ListstreamByExample(AuthorsExampleexample);//以stream形式从mysql获取数据
}

Mapper xml文件核心片段,以下两条select的唯一差异就是在stream获取数据的方式中多了一条属性: fetchSize="-2147483648"

<selectid="selectByExample"parameterType="com.alphathur.mysqlstreamingexport.domain.AuthorsExample"resultMap="BaseResultMap">
select
<iftest="distinct">
distinct
if>
'false'asQUERYID,
<includerefid="Base_Column_List"/>
fromauthors
<iftest="_parameter!=null">
<includerefid="Example_Where_Clause"/>
if>
<iftest="orderByClause!=null">
orderby${orderByClause}
if>
select>
<selectid="streamByExample"fetchSize="-2147483648"parameterType="com.alphathur.mysqlstreamingexport.domain.AuthorsExample"resultMap="BaseResultMap">
select
<iftest="distinct">
distinct
if>
'false'asQUERYID,
<includerefid="Base_Column_List"/>
fromauthors
<iftest="_parameter!=null">
<includerefid="Example_Where_Clause"/>
if>
<iftest="orderByClause!=null">
orderby${orderByClause}
if>
select>

获取数据的核心service如下,由于只做个简单演示,就懒得写成接口了。其中 streamDownload 方法即为stream取数据写文件的实现,它将以很低的内存占用从MySQL获取数据;此外还提供traditionDownload方法,它是一种传统的下载方式,批量获取全部数据,然后将每个对象写入文件。

@Service
publicclassAuthorsService{
privatefinalSqlSessionTemplatesqlSessionTemplate;
privatefinalAuthorsMapperauthorsMapper;

publicAuthorsService(SqlSessionTemplatesqlSessionTemplate,AuthorsMapperauthorsMapper){
this.sqlSessionTemplate=sqlSessionTemplate;
this.authorsMapper=authorsMapper;
}

/**
*stream读数据写文件方式
*@paramhttpServletResponse
*@throwsIOException
*/
publicvoidstreamDownload(HttpServletResponsehttpServletResponse)
throwsIOException{
AuthorsExampleauthorsExample=newAuthorsExample();
authorsExample.createCriteria();
HashMapparam=newHashMap<>();
param.put("oredCriteria",authorsExample.getOredCriteria());
param.put("orderByClause",authorsExample.getOrderByClause());
CustomResultHandlercustomResultHandler=newCustomResultHandler(newDownloadProcessor(httpServletResponse));
sqlSessionTemplate.select(
"com.alphathur.mysqlstreamingexport.mapper.AuthorsMapper.streamByExample",param,customResultHandler);
httpServletResponse.getWriter().flush();
httpServletResponse.getWriter().close();
}

/**
*传统下载方式
*@paramhttpServletResponse
*@throwsIOException
*/
publicvoidtraditionDownload(HttpServletResponsehttpServletResponse)
throwsIOException{
AuthorsExampleauthorsExample=newAuthorsExample();
authorsExample.createCriteria();
Listauthors=authorsMapper.selectByExample(authorsExample);
DownloadProcessordownloadProcessor=newDownloadProcessor(httpServletResponse);
authors.forEach(downloadProcessor::processData);
httpServletResponse.getWriter().flush();
httpServletResponse.getWriter().close();
}
}

下载的入口controller:

@RestController
@RequestMapping("download")
publicclassHelloController{
privatefinalAuthorsServiceauthorsService;

publicHelloController(AuthorsServiceauthorsService){
this.authorsService=authorsService;
}

@GetMapping("streamDownload")
publicvoidstreamDownload(HttpServletResponseresponse)
throwsIOException{
authorsService.streamDownload(response);
}

@GetMapping("traditionDownload")
publicvoidtraditionDownload(HttpServletResponseresponse)
throwsIOException{
authorsService.traditionDownload(response);
}
}

实体类对应的表结构创建语句:

CREATETABLE`authors`(
`id`int(11)NOTNULLAUTO_INCREMENT,
`first_name`varchar(50)CHARACTERSETutf8COLLATEutf8_unicode_ciNOTNULL,
`last_name`varchar(50)CHARACTERSETutf8COLLATEutf8_unicode_ciNOTNULL,
`email`varchar(100)CHARACTERSETutf8COLLATEutf8_unicode_ciNOTNULL,
`birthdate`dateNOTNULL,
`added`timestampNOTNULLDEFAULTCURRENT_TIMESTAMP,
PRIMARYKEY(`id`)
)ENGINE=InnoDBAUTO_INCREMENT=10095DEFAULTCHARSET=utf8COLLATE=utf8_unicode_ci;

这里有个问题:如何短时间内创建大批量测试数据到MySQL呢?一种方式是使用存储过程 + 大杀器 select insert 语句!不太懂?

没关系,且看我另一篇文章 MySQL如何生成大批量测试数据 你就会明白了。如果你懒得看,我这里已经将生成的270多万条测试数据上传到网盘,你直接下载然后通过navicat导入就好了。

  • 链接:https://pan.baidu.com/s/1hqnWU2JKlL4Tb9nWtJl4sw
  • 提取码:nrp0

有了测试数据,我们就可以直接测试了。先启动项目,然后打开jdk bin目录下的 jconsole.exe

首先我们测试传统方式下载文件的内存占用,直接浏览器访问:http://localhost:8080/download/traditionDownload。

可以看出,下载开始前内存占用大概为几十M,下载开始后内存占用急速上升,峰值达到接近2.5G,即使是下载完成,堆内存也维持一个较高的占用,这实在是太可怕了,如果生产环境敢这么搞,不出意外肯定内存溢出。

edf9f144-c3b1-11ed-bfe3-dac502259ad0.png

接着我们测试stream方式文件下载的内存占用,浏览器访问:http://localhost:8080/download/streamDownload,当下载开始后,内存占用也会有一个明显的上升,但是峰值才到500M。对比于上面的方式,内存占用率足足降低了80%!怎么样,兴奋了吗!

ee0493e2-c3b1-11ed-bfe3-dac502259ad0.png

我们再通过记事本打开下载后的两个文件,发现内容没有缺斤少两,都是2727127行,完美!



审核编辑 :李倩


温馨提示:以上内容整理于网络,仅供参考,如果对您有帮助,留下您的阅读感言吧!
相关阅读
本类排行
相关标签
本类推荐

CPU | 内存 | 硬盘 | 显卡 | 显示器 | 主板 | 电源 | 键鼠 | 网站地图

Copyright © 2025-2035 诺佳网 版权所有 备案号:赣ICP备2025066733号
本站资料均来源互联网收集整理,作品版权归作者所有,如果侵犯了您的版权,请跟我们联系。

关注微信