項目中有幾個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做事情。以上純個人見解，若有不當之處，請不吝指出，共同進步！