行集的Java多线程处理

Question

我正在下拉一个充满数据的表，我需要处理这一点，并对每一行进行一点格式化，然后推出一个REST API。

我使用了一个PostgreSQL数据库和Java实现，其思想是将所有数据向下拉，获得行的数量，并向上旋转线程以一次处理一个块。

我已经建立了连接，并将表拉入缓存行集，并使用last()、getrow()和beforefirst()获取行计数。

我正试图找到一种方法，将行集的一大块拆分出来并将其交给处理，但我似乎看不到任何方法可以做到这一点。

有极限x和其他东西，但我想避免使用如此大小的数据进行大量的数据库调用。

任何想法都将不胜感激。

这就是我看到的东西

RowSet rst = RowSetProvider.newFactory().createCachedRowSet();
rst.setUrl(url);
rst.setUsername(username);
rst.setPassword(password);

String cmd = "select * from event_log";

rst.setCommand(cmd);
rst.execute();

ResultSetMetaData rsmd = rst.getMetaData();
int columnsNumber = rsmd.getColumnCount();

rst.last();
int size = rst.getRow();
int maxPerThread = 1000;
rst.beforeFirst();

int threadsToCreate = size / maxPerThread;

for (int loopCount = 0; loopCount < threadsToCreate; loopCount++)
{
    //Create chunk
    //Create thread
    //Pass chunk into thread and start it
    //Once chunk is finished then thread and chunk are destroyed
}

Answer 1

这是考虑JDBC交互的正确方法：

• 所有查询都像是一个特殊视图:选择foo,a和b之间的bar AS baz FROM foo INNER JOIN whatever；-这将有效地创建一个新的临时表。
• ResultSet是一个动态交互概念：ResultSet不是返回数据的转储。这就像是FileInputStream和磁盘上的文件之间的关系：ResultSet有一些方法为您提供数据，而这些数据可能是通过与数据库“live”聊天来获得的，以获得这些信息。ResultSet本身只有几个句柄，而不是实际数据，尽管它可能会进行一些缓存，但您不知道。

因此：

• ResultSet是完全不可并行的。如果您与多个线程共享一个ResultSet对象，则您编写了一个bug，并且无法从那里恢复。
• 在许多DBs中，'ask for the length‘等于是完全运行整个查询，因此非常慢。你可能不想那样做，从‘我想同时处理我接收到的信息’的角度来看，也没有真正的理由那样做。您选错了方法。
• 结果集可以（通常，出于性能原因，应该是！）html" target="_blank">配置为“forward only”，意思是：您可以通过调用.next()前进一行，一旦您这样做了，就不能返回。这大大降低了DB服务器的负载，因为它不必准备好正确响应跳回到起点的请求。

我建议你这么做：

• 您有一个“Controller”线程，它具有ResultSet并运行查询。
• 查询返回后，您就不知道有多少条记录了。但您确实知道您想要并行化的程度--您想要并发处理这些数据的线程数量。
• 因此，答案是：以ExecutorPool的形式旋转多个线程。然后，让您的控制器拉出行（调用ResultSet.Next()并通过调用所有不同的.getFoo(idxOrColName)方法将所有数据拉入Java类型），将所有数据封送到单个Java对象中。我建议您编写一个代表一行数据的POJO并为每一行创建一个。
• 然后，您的控制器线程将接受此对象，并将此对象视为“作业”。

现在，您已经将问题简化为一个基本的forkjoin风格策略：您有一个线程来生成作业，您有一些代码来接受一个作业并完成它。我刚刚描述了ExecutorPool和friends的设计目的。

处理器线程不能访问ResultSet对象，这一点至关重要。从DB并行提取行是没有意义的，因为DB不是并行的，不能比单线程更快地提供这些信息。在这里，您能够获得的唯一并行化胜利是以并发的方式处理数据，这就是为什么如果不进行更大的修改，就无法改进上面的模型。

如果你正在寻找大刀阔斧的重新设计，你需要‘预分块’。比方说，您已经知道有一个包含一百万行的数据库，并且每一行都具有一个完全随机的ID的属性。您还知道您有X个处理器线程，其中X是一个动态数字，取决于许多因素，例如您运行的硬件有多少CPU核。

然后：

启动X线程。您告诉每个线程它的索引（因此，如果您有7个线程，一个线程有“索引0”，另一个线程有“索引1”，一直到“索引6”），以及总共有多少个线程。

然后，每个线程运行以下查询：

从unid%7=5的作业中选择*；

这是第6个作业线程将运行的查询。

这保证了每个线程运行的作业数量相等。

一般来说，这比前面的模型效率要低，因为这很可能意味着DB只是在做更多的工作（运行相同的查询7次，而不是只运行一次），并且任何给定的工作线程都可能在其他线程仍在运行时开始空闲，而不是在Controller-that-pulls-hands-jobs-out模型中，您不会遇到一个线程已经完成而其他线程仍有大量剩余作业的情况。

注意：RowSet和ResultSet以完全相同的方式有效地工作。实际上，RowSet的DB版本（JdbcRowSet）是作为ResultSet的轻型包装器实现的。