2

我想解析文件并将内容传输到数据库中。为了加快一切，文件应该被并行解析。
我有一个主线程，它逐行读取文件并创建Runnable，将其提供给ThreadPoolExecutor。每个Runnable都有自己的Session。

每行都包含一个客户端的唯一标识符，因此可以进行重复。系统尝试通过标识符在数据库中查找客户端。
如果找不到想要相同客户端的线程之一，则需要创建客户端。我在这里有一个“加入”点，其他线程必须等待允许创建客户端的线程。 

c = (Client) s.get("Client", identfier);
if (c == null) {
    CountDownLatch lock = isClientResolutionActive(identfier);

    if (lock != null) {
        lock.await();
        LOGGER.info("Lock was released ... " + identfier);
        c = (Client) s.get("Client", identfier);
    }
}

if (c == null) {
    c = createClient(...);

    s.save(c);
    s.flush();
    removeClientResolutionActive(identfier);
}

为了同步它们，我在调用者类中创建了两种方法，一种方法专用于检查是否已经有人在创建客户端并返回共享对象，另一种方法从列表中删除条目并通知所有等待线程。

我在互联网上搜索了很多，并试图找到我的问题或类似的问题，但没有成功。
此外，我不确定应该使用哪个并发对象。经过研究，我决定使用CountDownLatch。它用1初始化。应该只有一个线程创建它。（也许使用CountDownLatch以外的其他东西会更好，但我不知道是什么）

上述方法在地图上包含一个同步块，其中包含客户端的标识符和CountDownLatch的实例。

private CountDownLatch isClientResolutionActive(String identfier) {
    synchronized (activeSearches) {
        if (activeSearches.containsKey(identfier)) {
            // Only create the CountDownLatch if there are multiple threads for
            // that identfier
            if (activeSearches.get(identfier) == null) {
                activeSearches.put(identfier, new CountDownLatch(1));
            }
            return activeSearches.get(identfier);
        } else {
            LOGGER.info("Locked " + identfier);
            activeSearches.put(identfier, null);
            return null;
        }
    }
}

private void removeClientResolutionActive(String identfier) {
    synchronized (activeSearches) {
        CountDownLatch cl = activeSearches.get(identfier);
        activeSearches.remove(identfier);
        if (cl != null) {
            LOGGER.info("Unlock " + identfier);
            cl.countDown();
        }
    }
}

一般来说它工作正常，但有时我会遇到问题，当锁存器被释放（并删除）并且访问同步变量队列包含另一个线程来搜索已删除的条目（检查是否有任何线程已经在执行它)，它会尝试再次创建一个新客户端。 

18:02:55,611 [pool-1-thread-2] INFO LogImporter Unlock b42fcae346fbb2b1e3c544fb816de2c5
18:02:55,611 [pool-1-thread-3] INFO LogImporter Locked b42fcae346fbb2b1e3c544fb816de2c5
18:02:55,611 [pool-1-thread-4] INFO LogImporter Lock was released ... b42fcae346fbb2b1e3c544fb816de2c5

我想我必须改进同步，但我不知道如何。

一种想法是将客户端搜索移动到同步块中，或者在再次锁定数据库之前进行检查。
也许创建一个缓存或映射，其中包含数据库中所有已知的客户端。
或者在应用程序的整个生命周期中只使用一个会话？

提前感谢您的任何建议和提示。

4

1 回答 1

2

在并行线程中解析相同的文件不会增加速度，只会消耗额外的资源

一个问题较少且效率更高的 text2db 优化包括：

  • 批量读取文件（而不是逐行读取 1 MB，处理它，读取下一个 MB）

  • 批量插入数据库 - mysql 像这样：

    insert into urtable 
values
('val1','val2'),
('val1','val2');

（示例从http://bytes.com/topic/sql-server/answers/585793-insert-into-using-select-values-inserting-multiple-rows窃取- 抱歉懒得自己编一个）

  • 尝试防止 sql 来回运行（意味着：如果需要从数据库中选择输出来丰富您的数据集，请预先读取它，而不是在遍历文件时不断地读取它）

更新 - -

根据我的评论，在解析文件时可能需要从数据库中获取数据。好吧，如果你必须做，你必须做。但是：尽量不要这样做。

首先：读取具体数据可以看缓存与否。在狭隘的理解中，缓存只是通过任何启发式方法将磁盘数据移动到内存（不知道发生了什么）。我个人会尽量避免这种情况，因为启发式可能会与您对立。在更广泛的理解中，缓存是我在 PLUS 将数据从磁盘存储到内存之前所描述的，您可以精确定位（例如，通过 ID 或任何过滤条件）。所以我仍然不喜欢这种狭隘的理解部分，而是预先选择明确定义的数据的行为。

其次：我的个人经历是这样的：如果您正在处理文件解析中的完全规范化的数据模型数据库读取操作，通常会归结为“给我主键”我之前转储到数据库中的内容。当您一次写入多行时，这似乎变得很棘手。但是，尤其是在 MySQL 中，您绝对可以依赖“每个插入语句（甚至多行插入）都是原子的”，您可以从 last_insert_id() 获得 ID，因此您可以将其追踪到您之前编写的所有记录。我很确定其他数据库系统也有类似的“失败”。

第三：解析 LARGE 文件是我会尝试作为一项工作来操作，只有一个技术用户触发该工作，并确保这些进程中不超过 1 个并行运行。否则，您需要解决从文件锁定进入会话权限读/写管理开始的各种问题。因此，将其作为作业运行（至少在我的个人政策中）将其分类为分配大量 RAM - 取决于成本和速度的重要性。这意味着我什至不会费心将 100 K 行的关键字到 ID 表加载到内存中。

于 2013-10-31T11:39:53.740 回答