Java jdbc Sql Server并行写入

我的总体想法是为每个目标表(本例中为6个)生成一个新线程,并为每个表创建一个不同的jdbc连接,因此从理论上讲,每个写操作都可以并行地、独立地完成。

对我来说,这当然是个好计划。我会使用一个连接池,比如 HikariCP 或 DBCP 以保持与数据库服务器的多个连接。然后可以添加多个线程,每个线程都可以请求一个连接,然后将其返回到池中供以后使用。

这样行吗?其他/更好方式的建议?

它会起作用的。有一件事要考虑的是6可能不是正确的数字。您的服务器可能没有足够的带宽同时处理大量的数据,因此您可能需要考虑减少池中的线程数,直到找到最佳的数字,这将给您带来最大的带宽。也就是说,如果有6个表,那么6可能确实是正确的数字,这取决于数据在服务器上的分区方式。

根据您对线程的了解程度,您应该查看 docs on thread pooling .

我已经实现了以下解决方案,它使用producer/consumer模式 BlockingQueue 和 ExecutorService . 主线程(producer)为每个工作线程(consumer)实例化一个BlockingQueue,以及一个布尔易失性变量“terminated”,以在所有数据生成时向工作线程发出信号,并且它们应该终止执行(从while循环中转义,清空队列,并在jdbc连接上写入剩余数据)。生产者使用两个BlockingQueue blockingQueue1和blockingQueue2为每个线程生成不同的数据。

下面是简化的MainThreadProducer,它只为两个工作线程生成整数数据:

// MainThreadProducer.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class MainThreadProducer {

    public static Logger logger = LogManager.getLogger(MainThreadProducer.class);
    public final static BlockingQueue<Integer> blockingQueue1 = new LinkedBlockingDeque<>(100);
    public final static BlockingQueue<Integer> blockingQueue2 = new LinkedBlockingDeque<>(100);

    /* signal to the worker threads that all data has been generated */
    public static volatile boolean terminated = false;


    private void run () {

        try {

            ExecutorService executor = Executors.newFixedThreadPool(2);
            Future<Integer> future1 = executor.submit(new WorkerThreadConsumer("1"));
            Future<Integer> future2 = executor.submit(new WorkerThreadConsumer("2"));

            for (int i = 0; i < 10023; ++i) {

                blockingQueue1.put(i);
                blockingQueue2.put(i*2);

            }

            executor.shutdown();
            terminated = true;

            int res1 = future1.get();
            int res2 = future1.get();

            logger.info("Total rows written (thread 1): " + res1);
            logger.info("Total rows written (thread 2): " + res2);            

        }
        catch (Exception e) {

            e.printStackTrace();
            System.exit(1);

        }

    }

    public static void main(String[] args) {

        MainThreadProducer instance = new MainThreadProducer();
        instance.run();

    }

}

这是WorkerThreadConsumer。java类。对于这个测试,我将创建两个线程,分别在表TARGET_1和TARGET_2上写入数据库DBTEST。 每个线程都用特定的字符串类型(1和2)实例化,因此它可以知道需要从哪个BlockingQueue读取数据。

// WorkerThreadConsumer.java

import java.sql.PreparedStatement;
import com.microsoft.sqlserver.jdbc.SQLServerResultSet;
import java.sql.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import Configuration;


public class WorkerThreadConsumer implements Callable<Integer> {

    private String type;

    public WorkerThreadConsumer (String type) {

        this.type = type;        

    }

    @Override
    public Integer call() {        

        String TAG = "[THREAD_" + Thread.currentThread().getId() + "]";
        int processed = 0; // number of rows currently processed
        int batchSize = 100; // size of the batch we write to the server with the PreparedStatement

        try {
            // load jdbc driver
            Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver");
            MainThreadProducer.logger.info(TAG + "\tLoaded com.microsoft.sqlserver.jdbc.SQLServerDriver");

            String stub = String.format("INSERT INTO DBTEST.dbo.TARGET_%s (id) VALUES (?);", this.type);

            BlockingQueue<Integer> queue;

            switch (this.type) {
                case "1":
                    queue = MainThreadProducer.blockingQueue1;
                    break;

                case "2":
                    queue = MainThreadProducer.blockingQueue2;
                    break;

                default:
                    queue = MainThreadProducer.blockingQueue1;
            }

            try (Connection connection = DriverManager.getConnection(Configuration.DWH_DB_CONNECTION_URL);
                 PreparedStatement stmt = connection.prepareStatement(stub);) {

                connection.setAutoCommit(false);

                while (!MainThreadProducer.terminated) {

                    int data = queue.take();
                    stmt.setInt(1, data);
                    stmt.addBatch();
                    processed += 1;

                    if (processed % batchSize == 0) {
                        int[] result = stmt.executeBatch();
                        connection.commit();
                        MainThreadProducer.logger.info(TAG + "\tWritten rows count: " + result.length);
                    }

                }

                // empty queue and write
                while (!queue.isEmpty()) {
                    int data = queue.take();
                    stmt.setInt(1, data);
                    stmt.addBatch();
                    processed += 1;

                    if (processed % batchSize == 0) {
                        int[] result = stmt.executeBatch();
                        connection.commit();
                        MainThreadProducer.logger.info(TAG + "\tWritten rows count: " + result.length);
                    }
                }

                // last write in case queue size > batch size
                int[] result = stmt.executeBatch();
                connection.commit();
                MainThreadProducer.logger.info(TAG + "\tWritten rows count: " + result.length);


            }

        }
        catch (Exception e) {

            e.printStackTrace();
            System.exit(1);

        }

        return processed;
    }



}

这个解决方案似乎奏效了。如果你发现潜在问题,请告诉我。