springboot-mybatis/JPA流式查询的多种实现方式
Fire_Bit 人气:0项目中有几个batch需要检查所有的用户参与的活动的状态,以前是使用分页,一页一页的查出来到内存再处理,但是随着数据量的增加,效率越来越低。于是经过一顿搜索,了解到流式查询这么个东西,不了解不知道,这一上手,爱的不要不要的,效率贼高。项目是springboot 项目,持久层用的mybatis,整好mybatis的版本后,又研究了一下JPA的版本,做事做全套,最后又整了原始的JDBCTemplate 版本。废话不多说,代码如下:
第一种方式: springboot + mybatis 流式查询(网上说的有三种,我觉得下面这种最简单,对业务代码侵入性最小)
a) service 层代码:
package com.example.demo.service; import com.example.demo.bean.CustomerInfo; import com.example.demo.mapper.UserMapper; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.cursor.Cursor; import org.springframework.context.ApplicationContext; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import javax.annotation.Resource; @Slf4j @Service public class TestStreamQueryService { @Resource private ApplicationContext applicationContext; @Resource private UserMapper userMapper; @Resource private JdbcTemplate jdbcTemplate; @Transactional public void testStreamQuery(Integer status) { mybatisStreamQuery(status); } private void mybatisStreamQuery(Integer status) { log.info("waiting for query....."); Cursor<CustomerInfo> customerInfos = userMapper.getCustomerInfo(status); log.info("finish query!"); for (CustomerInfo customerInfo : customerInfos) { //处理业务逻辑 log.info("===============>{}", customerInfo.getId()); } } }
需要注意的有两点:
1.是userMapper 返回的是一个Cursor类型,其实就是用游标。然后遍历这个cursor,mybatis就会按照你在userMapper里设置的fetchSize 大小,每次去从数据库拉取数据
2.注意 testStreamQuery 方法上的 @transactional 注解,这个注解是用来开启一个事务,保持一个长连接(就是为了保持长连接采用的这个注解),因为是流式查询,每次从数据库拉取固定条数的数据,所以直到数据全部拉取完之前必须要保持连接状态。(顺便提一下,如果说不想让在这个testStreamQuery 方法内处理每条数据所作的更新或查询动作都在这个大事务内,那么可以另起一个方法 使用required_new 的事务传播,使用单独的事务去处理,使事务粒度最小化。如下图:)
b) mapper 层代码:
package com.example.demo.mapper; import com.example.demo.bean.CustomerInfo; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.cursor.Cursor; import org.springframework.stereotype.Repository; @Mapper @Repository public interface UserMapper { Cursor<CustomerInfo> getCustomerInfo(Integer status); }
mapper.xml
<?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> <mapper namespace="com.example.demo.mapper.UserMapper"> <select id="getCustomerInfo" resultType="com.example.demo.bean.CustomerInfo" fetchSize="2" resultSetType="FORWARD_ONLY"> select * from table_name where status = #{status} order by id </select> </mapper>
UserMapper.java 无需多说,其实要注意的是mapper.xml中的配置:fetchSize 属性就是上一步说的,每次从数据库取多少条数据回内存。resultSetType属性需要设置为 FORWARD_ONLY, 意味着,查询只会单向向前读取数据,当然这个属性还有其他两个值,这里就不展开了。
至此,springboot+mybatis 流式查询就可以用起来了,以下是执行结果截图:
c)读取200万条数据,每次fetchSize读取1000条,batch总用时50s左右执行完,速度是相当可以了,堆内存占用不超过250M,这里用的数据库是本地docker起的一个postgre, 远程数据库的话,耗时可能就不太一样了
第二种方式:springboot+JPA 流式查询
a) service层代码:
package com.example.demo.service; import com.example.demo.dao.CustomerInfoDao; import com.example.demo.mapper.UserMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationContext; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import javax.annotation.Resource; import javax.persistence.EntityManager; import java.util.stream.Stream; @Slf4j @Service public class TestStreamQueryService { @Resource private ApplicationContext applicationContext; @Resource private UserMapper userMapper; @Resource private JdbcTemplate jdbcTemplate; @Resource private CustomerInfoDao customerInfoDao; @Resource private EntityManager entityManager; @Transactional(readOnly = true) public void testStreamQuery(Integer status) { jpaStreamQuery(status); } public void jpaStreamQuery(Integer status) { Stream<com.example.demo.entity.CustomerInfo> stream = customerInfoDao.findByStatus(status); stream.forEach(customerInfo -> { entityManager.detach(customerInfo); //解除强引用,避免数据量过大时,强引用一直得不到GC 慢慢会OOM log.info("====>id:[{}]", customerInfo.getId()); }); } }
注意点:1. 这里的@transactional(readonly=true) 这里的作用也是保持一个长连接的作用,同时标注这个事务是只读的。
2. 循环处理数据时需要先:entityManager.detach(customerInfo); 解除强引用,避免数据量过大时,强引用一直得不到GC 慢慢会OOM。
b) dao层代码:
package com.example.demo.dao; import com.example.demo.entity.CustomerInfo; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.QueryHints; import org.springframework.stereotype.Repository; import javax.persistence.QueryHint; import java.util.stream.Stream; import static org.hibernate.jpa.QueryHints.HINT_FETCH_SIZE; @Repository public interface CustomerInfoDao extends JpaRepository<CustomerInfo, Long> { @QueryHints(value=@QueryHint(name = HINT_FETCH_SIZE,value = "1000")) Stream<CustomerInfo> findByStatus(Integer status); }
注意点:1.dao方法的返回值是 Stream 类型
2.dao方法的注解:@QueryHints(value=@QueryHint(name = HINT_FETCH_SIZE,value = "1000")) 这个注解是设置每次从数据库拉取多少条数据,自己可以视情况而定,不可太大,反而得不偿失,一次读取太多数据数据库也是很耗时间的。。。
自此springboot + jpa 流式查询代码就贴完了,可以happy了,下面是执行结果:
c) batch读取两百万条数据,堆内存使用截图:
每次fetchSize拉取1000条数据,可以看到内存使用情况:初始内存不到100M,batch执行过程中最高内存占用300M出头然后被GC。读取效率:不到一分钟执行完(处理每一条数据只是打印一下id),速度还是非常快的。
d) 读取每一条数据时,不使用 entityManager.detach(customerInfo),内存使用截图:
最终OOM了,这里的entityManager.detach(customerInfo) 很关键。
第三种方式:使用JDBC template 流式查询
其实这种方式就是最原始的jdbc的方式,代码侵入性很大,逼不得已也不会使用
a) 上代码:
package com.example.demo.service; import com.example.demo.dao.CustomerInfoDao; import com.example.demo.mapper.UserMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationContext; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import javax.persistence.EntityManager; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; @Slf4j @Service public class TestStreamQueryService { @Resource private ApplicationContext applicationContext; @Resource private UserMapper userMapper; @Resource private JdbcTemplate jdbcTemplate; @Resource private CustomerInfoDao customerInfoDao; @Resource private EntityManager entityManager; public void testStreamQuery(Integer status) { jdbcStreamQuery(status); } private void jdbcStreamQuery(Integer status) { Connection conn = null; PreparedStatement pstmt = null; ResultSet rs = null; try { conn = jdbcTemplate.getDataSource().getConnection(); conn.setAutoCommit(false); pstmt = conn.prepareStatement("select * from customer_info where status = " + status + " order by id", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); pstmt.setFetchSize(1000); pstmt.setFetchDirection(ResultSet.FETCH_FORWARD); rs = pstmt.executeQuery(); while (rs.next()) { long id = rs.getLong("id"); String name = rs.getString("name"); String email = rs.getString("email"); int sta = rs.getInt("status"); log.info("=========>id:[{}]", id); } } catch (SQLException throwables) { throwables.printStackTrace(); } finally { try { rs.close(); pstmt.close(); conn.close(); } catch (SQLException throwables) { throwables.printStackTrace(); } } } }
b) 执行结果:200万数据不到50秒执行完,内存占用最高300M
自此,针对不同的持久层框架, 使用不同的流式查询,其实本质是一样的,归根结底还是驱动jdbc做事情。以上纯个人见解,若有不当之处,请不吝指出,共同进步!
加载全部内容