springboot-mybatis/JPA流式查詢的多種實現(xiàn)方式

更新時間：2022年12月19日 08:24:06 作者：Fire_Bit

這篇文章主要介紹了springboot-mybatis/JPA流式查詢,本文給大家分享三種方式，每種方式結合示例代碼給大家講解的非常詳細，需要的朋友可以參考下

項目中有幾個batch需要檢查所有的用戶參與的活動的狀態(tài)，以前是使用分頁，一頁一頁的查出來到內(nèi)存再處理，但是隨著數(shù)據(jù)量的增加，效率越來越低。于是經(jīng)過一頓搜索，了解到流式查詢這么個東西，不了解不知道，這一上手，愛的不要不要的，效率賊高。項目是springboot 項目，持久層用的mybatis，整好mybatis的版本后，又研究了一下JPA的版本，做事做全套，最后又整了原始的JDBCTemplate 版本。廢話不多說，代碼如下：

第一種方式： springboot + mybatis 流式查詢（網(wǎng)上說的有三種，我覺得下面這種最簡單，對業(yè)務代碼侵入性最?。?/h2>

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) {
            //處理業(yè)務邏輯
            log.info("===============>{}", customerInfo.getId());
        }
    }
}

需要注意的有兩點：

1.是userMapper 返回的是一個Cursor類型，其實就是用游標。然后遍歷這個cursor，mybatis就會按照你在userMapper里設置的fetchSize 大小，每次去從數(shù)據(jù)庫拉取數(shù)據(jù)

2.注意 testStreamQuery 方法上的 @transactional 注解，這個注解是用來開啟一個事務，保持一個長連接（就是為了保持長連接采用的這個注解），因為是流式查詢，每次從數(shù)據(jù)庫拉取固定條數(shù)的數(shù)據(jù)，所以直到數(shù)據(jù)全部拉取完之前必須要保持連接狀態(tài)。（順便提一下，如果說不想讓在這個testStreamQuery 方法內(nèi)處理每條數(shù)據(jù)所作的更新或查詢動作都在這個大事務內(nèi)，那么可以另起一個方法使用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 屬性就是上一步說的，每次從數(shù)據(jù)庫取多少條數(shù)據(jù)回內(nèi)存。resultSetType屬性需要設置為 FORWARD_ONLY，意味著，查詢只會單向向前讀取數(shù)據(jù)，當然這個屬性還有其他兩個值，這里就不展開了。

至此，springboot+mybatis 流式查詢就可以用起來了，以下是執(zhí)行結果截圖：

c）讀取200萬條數(shù)據(jù)，每次fetchSize讀取1000條，batch總用時50s左右執(zhí)行完，速度是相當可以了，堆內(nèi)存占用不超過250M，這里用的數(shù)據(jù)庫是本地docker起的一個postgre, 遠程數(shù)據(jù)庫的話，耗時可能就不太一樣了

第二種方式：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); //解除強引用，避免數(shù)據(jù)量過大時，強引用一直得不到GC 慢慢會OOM
            log.info("====>id:[{}]", customerInfo.getId());
        });
    }
 
}

注意點：1. 這里的@transactional（readonly=true）這里的作用也是保持一個長連接的作用，同時標注這個事務是只讀的。

2. 循環(huán)處理數(shù)據(jù)時需要先：entityManager.detach(customerInfo); 解除強引用，避免數(shù)據(jù)量過大時，強引用一直得不到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")) 這個注解是設置每次從數(shù)據(jù)庫拉取多少條數(shù)據(jù)，自己可以視情況而定，不可太大，反而得不償失，一次讀取太多數(shù)據(jù)數(shù)據(jù)庫也是很耗時間的。。。

自此springboot + jpa 流式查詢代碼就貼完了，可以happy了，下面是執(zhí)行結果：

c) batch讀取兩百萬條數(shù)據(jù)，堆內(nèi)存使用截圖：

每次fetchSize拉取1000條數(shù)據(jù)，可以看到內(nèi)存使用情況：初始內(nèi)存不到100M，batch執(zhí)行過程中最高內(nèi)存占用300M出頭然后被GC。讀取效率：不到一分鐘執(zhí)行完（處理每一條數(shù)據(jù)只是打印一下id），速度還是非?？斓摹?/p>

d) 讀取每一條數(shù)據(jù)時，不使用 entityManager.detach(customerInfo),內(nèi)存使用截圖：

最終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) 執(zhí)行結果：200萬數(shù)據(jù)不到50秒執(zhí)行完，內(nèi)存占用最高300M