HikariCP 的 Postgresql 性能问题

Question

我正在尝试以小批量（每个 csv 中 6000 行）将大数据加载到 PostgreSQL 服务器中的一个 table（总共 4000 万行）。我认为 HikariCP 非常适合这个目的。

这是我从数据插入中获得的吞吐量 Java 8 (1.8.0_65)，Postgres JDBC 驱动程序 9.4.1211 和 HikariCP 2.4.3.

4 分 42 秒 6000 行。

我做错了什么，如何提高插入速度？

关于我的设置的更多信息：

程序在公司网络后面的笔记本电脑中运行。
Postgres 服务器 9.4 是带有数据库的 Amazon RDS。m4.large 和 50 GB SSD。
尚未在 table 上定义明确的索引或主键。

程序使用大线程池异步插入每一行以保存请求，如下所示：

private static ExecutorService executorService = new ThreadPoolExecutor(5, 1000, 30L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(100000));

数据源配置为：

        private DataSource getDataSource() {
                if (datasource == null) {
                    LOG.info("Establishing dataSource");
                    HikariConfig config = new HikariConfig();
                    config.setJdbcUrl(url);
                    config.setUsername(userName);
                    config.setPassword(password);
                    config.setMaximumPoolSize(600);// M4.large 648 connections tops
                    config.setAutoCommit(true); //I tried autoCommit=false and manually committed every 1000 rows but it only increased 2 minute and half for 6000 rows
                    config.addDataSourceProperty("dataSourceClassName","org.postgresql.ds.PGSimpleDataSource");
                    config.addDataSourceProperty("dataSource.logWriter", new PrintWriter(System.out));
                    config.addDataSourceProperty("cachePrepStmts", "true");
                    config.addDataSourceProperty("prepStmtCacheSize", "1000");
                    config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
                    config.setConnectionTimeout(1000);

                    datasource = new HikariDataSource(config);
                }
                return datasource;
            }

这是我阅读源数据的地方：

    private void readMetadata(String inputMetadata, String source) {
            BufferedReader br = null;
            FileReader fr = null;
            try {
                br = new BufferedReader(new FileReader(inputMetadata));
                String sCurrentLine = br.readLine();// skip header;
                if (!sCurrentLine.startsWith("xxx") && !sCurrentLine.startsWith("yyy")) {
                    callAsyncInsert(sCurrentLine, source);
                }
                while ((sCurrentLine = br.readLine()) != null) {
                    callAsyncInsert(sCurrentLine, source);
                }
            } catch (IOException e) {
                LOG.error(ExceptionUtils.getStackTrace(e));
            } finally {
                try {
                    if (br != null)
                        br.close();

                    if (fr != null)
                        fr.close();

                } catch (IOException ex) {
                    LOG.error(ExceptionUtils.getStackTrace(ex));
                }
            }
    }

我正在异步插入数据（或尝试使用 jdbc！）：

            private void callAsyncInsert(final String line, String source) {
                    Future<?> future = executorService.submit(new Runnable() {
                        public void run() {
                            try {
                                dataLoader.insertRow(line, source);
                            } catch (SQLException e) {
                                LOG.error(ExceptionUtils.getStackTrace(e));
                                try {
                                    errorBufferedWriter.write(line);
                                    errorBufferedWriter.newLine();
                                    errorBufferedWriter.flush();
                                } catch (IOException e1) {
                                    LOG.error(ExceptionUtils.getStackTrace(e1));
                                }
                            }
                        }
                    });
                    try {
                        if (future.get() != null) {
                            LOG.info("$$$$$$$$" + future.get().getClass().getName());
                        }
                    } catch (InterruptedException e) {
                        LOG.error(ExceptionUtils.getStackTrace(e));
                    } catch (ExecutionException e) {
                        LOG.error(ExceptionUtils.getStackTrace(e));
                    }
                }

我的DataLoader.insertRow如下：

            public void insertRow(String row, String source) throws SQLException {
                    String[] splits = getRowStrings(row);
                    Connection conn = null;
                    PreparedStatement preparedStatement = null;
                    try {
                        if (splits.length == 15) {
                            String ... = splits[0];
                            //blah blah blah

                            String insertTableSQL = "insert into xyz(...) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ";
                            conn = getConnection();
                            preparedStatement = conn.prepareStatement(insertTableSQL);
                            preparedStatement.setString(1, column1);
                            //blah blah blah
                            preparedStatement.executeUpdate();
                            counter.incrementAndGet();
                            //if (counter.get() % 1000 == 0) {
                                //conn.commit();
                            //}
                        } else {
                            LOG.error("Invalid row:" + row);
                        }
                    } finally {
                        /*if (conn != null) {
                            conn.close();   //Do preparedStatement.close(); rather connection.close
                        }*/
                        if (preparedStatement != null) {
                            preparedStatement.close();
                        }
                    }
                }

在 pgAdmin4 中进行监控时，我注意到了一些事情：

最高每秒事务数接近 50。
活动数据库会话只有一个，会话总数为15。
块太多I/O（达到大约 500，不确定这是否应该是一个问题）

Answer 1

您绝对想使用批量插入，在循环外部准备语句，并关闭自动提交。在伪代码中：

PreparedStatement stmt = conn.prepareStatement("insert into xyz(...) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
while ( <data> ) {
   stmt.setString(1, column1);
   //blah blah blah
   stmt.addBatch();
}
stmt.executeBatch();
conn.commit();

即使是单个连接上的单个线程也应该能够插入 > 5000 rows/sec。

更新：如果你想多线程，连接数应该是数据库的数量CPU 核心 x1.5 或 2。处理线程的数量应该匹配，并且每个处理thread 应该使用上面的模式处理一个 CSV 文件。但是，您可能会发现许多并发插入同一个 table 会在数据库中造成过多的锁争用，在这种情况下，您需要减少处理线程的数量，直到找到最佳并发性。

适当大小的池和并发应该很容易超过 20K rows/sec。

另外，请升级到 HikariCP v2.6.0。

HikariCP 的 Postgresql 性能问题

Postgresql performance issue with HikariCP

postgresql

jdbc

hikaricp