JDBC读取数据的缺点

jdbc默认的读取数据的时候，会将要查询的数据一次性读到内存中，再通过resultSet循环读取出来，这样子百万条数据很容易就撑爆内存，再执行ResultSet的next方法，在数据全部返回之前，next方法是阻塞的，设置了之后（使用stream），ResultSet的next方法就会立即返回server端返回的数据，这样客户端就不用使用大量的内存存储返回的数据了
然后调研了下发现，其实可以通过jdbc流的方式读取数据，这种方式可以将读取出来的每一条数据查询到客户端，再执行insert操作，这样机器负载就会轻很多，实际操作了下，确实是这样的。

代码

  1. import com.alibaba.fastjson.JSONObject;
  2. import org.apache.flink.configuration.Configuration;
  3. import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
  5. import java.sql.*;
  6. import java.util.Properties;
  8. public class MysqlSource extends RichSourceFunction<JSONObject>{
  10.         private Properties properties;
  11.         private Connection connection;
  12.         PreparedStatement preparedStatement;
  14.         public MysqlSource() {
  15.             super();
  16.         }
  18.         public MysqlSource(Properties properties) {
  19.             this.properties = properties;
  20.         }
  22.         @Override
  23.         public void open(Configuration parameters) throws Exception {
  24.             super.open(parameters);
  25.             String jdbcUri = this.properties.getProperty("jdbcUri");
  26.             this.connection = DriverManager.getConnection(jdbcUri, "db_name", "passwrod");
  27.         }
  29.         @Override
  30.         public void run(SourceContext<JSONObject> ctx) throws Exception {
  31.             String sql = "select * from table_name";
  32.             // 表示流式读取数据
  33.             preparedStatement = this.connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
  34.             // 一定要设置成Integer.MIN_VALUE
  35.             preparedStatement.setFetchSize(Integer.MIN_VALUE);
  36.             ResultSet resultSet = preparedStatement.executeQuery();
  37.             while (resultSet.next()) {
  38.                 JSONObject jsonObject = new JSONObject();
  39.                 ctx.collect(jsonObject);
  40.             }
  41.         }
  42.         @Override
  43.         public void cancel() {
  44.             try {
  45.                 this.preparedStatement.close();
  46.                 this.connection.close();
  47.             } catch (SQLException e) {
  48.                 e.printStackTrace();
  49.             }
  50.         }
  51. }

具体使用stream来读取数据，只是需要在生成prepareStatement时加上ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY这两个参数，再设置一下FetchSize为Integer.MIN_VALUE就可以了，mysql判断是否开启流式读取结果就是看这三个条件forward-only，read-only，fatch size=Integer.MIN_VALUE有没有被设置
[

](https://blog.csdn.net/qq_22912803/article/details/88998415)