SpringBoot+Mybatis 如何实现流式查询,你知道吗?
点击上方“芋道源码”,选择“设为星标”
管她前浪,还是后浪?
能浪的浪,才是好浪!
每天 10:33 更新文章,每天掉亿点点头发...
源码精品专栏
前言
mybatis的流式查询,有点冷门,实际用的场景比较少,但是在某些特殊场景下,却是十分有效的一个方法。很多人没有听说过,实际上是对mybatis没有太重视,对mybatis想法还停留一个dao接口对应着mapper里的一个sql,mybatis的关键是如何写好sql以及sql的优化上;
其实mybatis远不止这些,通过这篇文章,和大家一块来见识一下流式查询,mybatis相对冷门的神秘面纱。文章的内容将从以下几个方面展开:
什么是mybatis的流式查询; Cursor接口的主要方法; 代码层面如何实现; 具体的应用场景; 使用中的一些注意事项;
基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址:https://github.com/YunaiV/ruoyi-vue-pro 视频教程:https://doc.iocoder.cn/video/
环境 配置
jdk版本:1.8 开发工具:Intellij iDEA 2020.1 springboot:2.3.9.RELEASE mybatis-spring-boot-starter:2.1.4
基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址:https://github.com/YunaiV/yudao-cloud 视频教程:https://doc.iocoder.cn/video/
什么是mybatis流式查询?
使用mybatis作为持久层的框架时,通过mybatis执行查询数据的请求执行成功后,mybatis返回的结果集不是一个集合或对象,而是一个迭代器,可以通过遍历迭代器来取出结果集,避免一次性取出大量的数据而占用太多的内存。
Cursor
org.apache.ibatis.cursor.Cursor
接口有三个抽象方法,分别是
isOpen() :判断cursor是否正处于打开状态; isConsumed() :判断查询结果是否全部读取完; getCurrentIndex() :查询已读取数据在全部数据里的索引位置;
public interface Cursor<T> extends Closeable, Iterable<T> {
//判断cursor是否正处于打开状态
//当返回true,则表示cursor已经开始从数据库里刷新数据了;
boolean isOpen();
//判断查询结果是否全部读取完;
//当返回true,则表示查询sql匹配的全部数据都消费完了;
boolean isConsumed();
//查询已读取数据在全部数据里的索引位置;
//第一条数据的索引位置为0;当返回索引位置为-1时,则表示已经没有数据可以读取;
int getCurrentIndex();
}
代码实现
mybatis的所谓流式查询,就是服务端程序查询数据的过程中,与远程数据库一直保持连接,不断的去数据库拉取数据,提交事务并关闭sqlsession后,数据库连接断开,停止数据拉取,需要注意的是使用这种方式,需要自己手动维护sqlsession和事务的提交。
1、实现方式很简单,原来返回的类型是集合或对象,流式查询返回的的类型Curor,泛型内表示实际的类型,其他没有变化;
@Mapper
public interface PersonDao {
Cursor<Person> selectByCursor();
Integer queryCount();
}
<select id="selectByCursor" resultMap="personMap">
select * from sys_person order by id desc
</select>
<select id="queryCount" resultType="java.lang.Integer">
select count(*) from sys_person
</select>
2、dao层向service层返回的是Cursor类型对象,只要不提交关闭sqlsession,服务端程序就可以一直从数据数据库读取数据,直到查询sql匹配到数据全部读取完;
示例里的主要业务逻辑是:从sys_person表中读取所有的人员信息数据,然后按照每1000条数据为一组,读取到内存里进行处理,以此类推,直到查询sql匹配到数据全部处理完,再提交事务,关闭sqlSession;
@Service
@Slf4j
public class PersonServiceImpl implements IPersonService {
@Autowired
private SqlSessionFactory sqlSessionFactory;
@Override
public void getOneByAsync() throws InterruptedException {
new Thread(new Runnable() {
@SneakyThrows
@Override
public void run() {
//使用sqlSessionFactory打开一个sqlSession,在没有读取完数据之前不要提交事务或关闭sqlSession
log.info("----开启sqlSession");
SqlSession sqlSession = sqlSessionFactory.openSession();
try {
//获取到指定mapper
PersonDao mapper = sqlSession.getMapper(PersonDao.class);
//调用指定mapper的方法,返回一个cursor
Cursor<Person> cursor = mapper.selectByCursor();
//查询数据总量
Integer total = mapper.queryCount();
//定义一个list,用来从cursor中读取数据,每读取够1000条的时候,开始处理这批数据;
//当前批数据处理完之后,清空list,准备接收下一批次数据;直到大量的数据全部处理完;
List<Person> personList = new ArrayList<>();
int i = 0;
if (cursor != null) {
for (Person person : cursor) {
if (personList.size() < 1000) {
// log.info("----id:{},userName:{}", person.getId(), person.getUserName());
personList.add(person);
} else if (personList.size() == 1000) {
++i;
log.info("----{}、从cursor取数据达到1000条,开始处理数据", i);
log.info("----处理数据中...");
Thread.sleep(1000);//休眠1s模拟处理数据需要消耗的时间;
log.info("----{}、从cursor中取出的1000条数据已经处理完毕", i);
personList.clear();
personList.add(person);
}
if (total == (cursor.getCurrentIndex() + 1)) {
++i;
log.info("----{}、从cursor取数据达到1000条,开始处理数据", i);
log.info("----处理数据中...");
Thread.sleep(1000);//休眠1s模拟处理数据需要消耗的时间;
log.info("----{}、从cursor中取出的1000条数据已经处理完毕", i);
personList.clear();
}
}
if (cursor.isConsumed()) {
log.info("----查询sql匹配中的数据已经消费完毕!");
}
}
sqlSession.commit();
log.info("----提交事务");
}catch (Exception e){
e.printStackTrace();
sqlSession.rollback();
}
finally {
if (sqlSession != null) {
//全部数据读取并且做好其他业务操作之后,提交事务并关闭连接;
sqlSession.close();
log.info("----关闭sqlSession");
}
}
}
}).start();
}
}
应用场景
其实mybatis的流式查询适用范围很有限,这里举个例子,假如有这样一个需求 :有50万员工的一年的工资数据明细,需要输出一张公司支出工资的数据报表。
需求很简单,估计有人是这样想:这太简单了,查询出员工的工资数据明细,然后按照套上公式逐条计算出结果,然后汇总计算结果,插入到新的结果表里不就行了。事实上这件事绝对不简单:
50万的数据全部读取到jvm的内存里得占用多大空间? 这么多对象的垃圾回收又需要多久? 这么多数据计算是高频行为还是低步行为? 如果计算到某条员工的数据发生异常,已经计算好的数据要不要全部回滚?...
总之,直接取出50万数据来计算,风险肯定不小。那怎么办呢?
在实际的开发中,也经常遇到一些百十万,说大不大,说小不小的数据报表处理,我的主要设计思路通常就是数据切隔+异步,具体怎么做呢?结合上面的例子,是这样的:
1、按照月份、省份或者部门,对工资明细数据进行数据切隔分组;
2、把不同月份、省份、部门的工资数据包装成多线程任务,放到线程池中去执行;
3、根据切隔的多线程任务数量,定义一个同步工具类CountDownLatch;
4、根据同步工具类CountDownLatch,来判断所有的多线程任务是否全部执行完;等到所有的多线程任务全部执行完成后,再执行汇总的逻辑;
5、在多线程任务里,查询具体月份、省份的员工工资数据明细的时候,如果数据量还是不少,就可以使用mybatis的流式查询,分批获取员工工资明细数据,进行当前批的计算、汇总,然后所有分批数据都计算完成后,再汇总所有分批数据;
注意事项
mybatis的流式查询的本意,是避免大量数据的查询而导致内存溢出,因此dao层查询返回的是一个迭代器(Cursor),可以每次从迭代器中取出一条查询结果,在实际业务开发过程中,即是根据实际的jvm内存大小,从迭代器中取出一定数量的数据后,再进行数据处理,待处理完之后,继续取出一定数据再处理,以此类推直到全部数据处理完,这样做的最大好处就是能够降低内存使用和垃圾回收器的负担,使数据处理的过程相对更加高效、可控,内存溢出的风险较小;
好处很明显,缺点也很就明显,处理的时间可能会变长,需要引入多线程异步操作,并且在迭代器遍历和数据处理的过程中,数据库连接不能断开,即当前sqlSession要保持持续打开状态,一量断开,数据读取就会中断,所以关于这块的处理,使用mybatis原生的sqlSession进行手动查询、提交事务、回滚和关闭sqlSession最为稳妥、最简单;
欢迎加入我的知识星球,一起探讨架构,交流源码。加入方式,长按下方二维码噢:
已在知识星球更新源码解析如下:
最近更新《芋道 SpringBoot 2.X 入门》系列,已经 101 余篇,覆盖了 MyBatis、Redis、MongoDB、ES、分库分表、读写分离、SpringMVC、Webflux、权限、WebSocket、Dubbo、RabbitMQ、RocketMQ、Kafka、性能测试等等内容。
提供近 3W 行代码的 SpringBoot 示例,以及超 4W 行代码的电商微服务项目。
获取方式:点“在看”,关注公众号并回复 666 领取,更多内容陆续奉上。
文章有帮助的话,在看,转发吧。
谢谢支持哟 (*^__^*)
微信扫码关注该文公众号作者