并发运行10万个进程

Question

我正在模拟一个银行系统，在这个系统中，我有10万笔交易要处理。每种类型的事务都实现runnable，并且我有可能发生的各种类型的事务。

transactions是一个可运行的数组。

理想情况下，以下代码将解决我的问题：

for (Transaction transaction : transactions) {
    new Thread(transaction).start();
}

但是，很明显，当尝试启动10万个线程时，一定会出现java.lang.OutOfMemoryError: unable to create new native thread。

接下来，我尝试实现一个ExecutorService来创建一个线程池来管理我的10万个可运行项。

ExecutorService service;
int cpus = Runtime.getRuntime().availableProcessors();
// cpus == 8 in my case
service = Executors.newFixedThreadPool(cpus);

for (Transaction transaction : transactions) {
    service.execute(transaction);
}

在尝试这种方法时，长进程会“占用”JVM。例如，一种类型的事务需要30-60秒才能执行。在分析应用程序时，在长事务发生时不允许运行其他线程。

​

​

在这种情况下，线程6在完成处理之前不允许任何其他线程运行。

因此，我的问题是:如何在不遇到内存问题的情况下尽可能快地运行10万个事务？如果ExecutorService是答案，那么我如何阻止非常长的事务占用JVM，并允许其他事务并发运行？

编辑：

我强迫某些类型的事务发生30 -60秒的故意，以确保我的线程程序正确工作。每个事务锁在一个帐户上，并且有10个帐户。下面是我的方法，它占据了JVM：(由run()调用)

public void makeTransaction() {
    synchronized(account) {
        long timeStarted = System.nanoTime();
        long timeToEnd = timeStarted + nanos;

        this.view = new BatchView(transactionNumber, account.getId());

        this.displayView();

        while(true) {
            if(System.nanoTime() % 1000000000 == 0) {
                System.out.println("batch | " + account.getId());
            }

            if(System.nanoTime() >= timeToEnd) {
                break;
            }
        }
    }
}

每次运行此事务时，只有一个帐户被锁定，剩下9个其他帐户可供处理。为什么JVM不再处理线程，而是挂起直到这个长事务完成？

下面是一个指向该项目的小型化版本的链接，以演示该问题：项目

Answer 1

应用程序的问题是，很快所有线程都会为同一个帐户分配事务，然后除了一个线程之外，所有线程都必须等待。如果我暂停了应用程序，您可以在下面的屏幕截图中看到这一点。线程池-1-线程-3目前正在用id 19处理帐户对象的事务(这个id不是您的帐户id，而是惟一的对象id Eclipse分配)，所有其他线程都在等待同一个帐户对象上的锁。帐户对象是您的id为9的对象。

​

​

这一切为什么要发生？在事务853中，一个线程启动第一个长时间运行的事务(用于帐户9)。其他线程继续处理其他事务。但是，当任何线程到达帐户9的另一个事务时，它将不得不停止等待。事务857、861和862也用于帐户9，每个线程阻塞一个线程，因此此时我的所有线程都被阻塞(在我的四核上)。

如何解决这个问题？这取决于您的用例。

如果在实际程序中保证给定帐户X没有传入事务，只要帐户X有另一个事务正在运行，则不需要更改任何内容。

如果您的帐户数量与线程数量相比非常大，那么问题就变得更加不可能了，所以您可能会决定接受它。

如果您的帐户数量相对较低(假设可能少于100个左右)，您应该(正如Peter所说)每个帐户有一个(无休止地运行)线程，每个线程都有自己的事务队列。这可能会更有效，因为线程不需要为共享队列“战斗”。

另一个解决办法是实施某种形式的“盗窃工作”。这意味着，每当线程被阻塞时，它都会寻找一些其他的工作来完成。要实现这一点，首先需要能够检查线程是否能够获得给定帐户的锁。在Java中使用synchronized是不可能的，所以您需要类似于ReentrantLock.tryLock()的东西。您还需要能够从每个线程直接访问事务队列，因此我想您不能在这里使用ExecutorService，而是需要实现自己处理事务(使用LinkedBlockingQueue)。

现在，每个线程将轮询循环中队列中的事务。首先，它试图通过tryLock()获取相应帐户的锁。如果失败，将事务添加到(特定于线程的)列表中，从队列中获取下一个事务，然后尝试执行此事务，直到找到您可以处理的事务为止。在事务完成后，首先查看列表中的当前情况--可能的话--在从全局队列中提取另一个事务之前处理事务。代码可以大致如下所示：

public BlockingQueue<Transaction> queue = ...; // the global queue for all threads

public void run() {
   LinkedList<Transaction> myTransactions = new LinkedList<>();
   while (true) {
     Transaction t = queue.take();
     while (!t.getLock().tryLock()) {
        myTransactions.add(t);
     }
     try {
       // here we hold the lock for t
       t.makeTransaction();
     } finally {
       t.getLock().unlock();
     }

     Iterator<Transaction> iter = myTransactions.iterator();
     while (iter.hasNext()) {
       t = iter.next();
       if (t.getLock().tryLock()) {
         try {
           t.makeTransaction();
         } finally {
           t.getLock().unlock();
         }
         iter.remove();
       }
     }
   }
 }

请注意，这至少还有以下问题需要解决：

• 当线程挂起在queue.take()中时，它不会检查其列表中的事务是否可用。因此，如果有一段时间queue为空(例如，在处理结束时)，则可能会有事务滞留在未处理的列表中。
• 如果一些线程持有大量锁，则剩余的线程可能会占用大量事务，它们现在无法处理这些事务，因此它们只会填充本地列表，耗尽全局队列。当释放锁时，许多事务可能已经从全局队列中删除，从而在线程可以完成的工作之间造成不平衡(一些线程可能在空闲，而其他线程仍在处理它们的长期事务积压)。

一个更简单的替代方法可能是，如果无法为队列获取锁，则可以将事务放入队列(在最后)，但这将使它们以非常任意的顺序执行(上面的解决方案也可能发生这种情况，但可能不会非常严重)。

编辑:更好的解决方案可能是在每个帐户上附加一个队列，而不是特定于线程的列表。然后，当线程发现该帐户被阻塞时，它会将事务添加到相应帐户的队列中。当线程完成帐户X的事务时，在查看全局列表之前，应该先查看帐户X的队列(如果在那里添加了任何事务)。

Answer 2

在分析应用程序时，在长事务发生时不允许运行其他线程。

最有可能的情况是，此任务使用的资源是单线程的。也就是说，写ti的方式可以防止并发使用。

如何在不遇到内存问题的情况下尽快运行100000个事务？

如果事务是CPU绑定的，那么您应该有一个与您拥有的CPU数量相同的池。

如果事务依赖于数据库，则应该考虑对它们进行批处理，以更有效地利用数据库。

如果ExecutorService是答案，那么我如何阻止非常长的事务占用JVM，并允许其他事务并发运行？

使交易变得更短。如果你的任务运行时间超过几毫秒，你应该弄清楚为什么要花这么长时间。首先，我将看看网络/IO是如何使用和分析任务的。大多数事务(如果您有大量的事务)应该在0.01秒左右或更少的理想情况下。

您应该非常小心地考虑如何使用共享资源。如果您的任务使用相同的资源太多，您可能会发现多线程不会更快，甚至更慢。

Answer 3

重要的是要计算工作线程的数量，它可以根据您的硬件并行地处理事务。几乎没有可用的公式来调整线程池的大小。

适用于CPU绑定应用程序

N*U或(N+1)*U

用于IO绑定应用程序

N*U* (1+W/C)

其中N -处理器数U -目标CPU利用率W -等待时间C -计算时间

例如，如果您的应用程序使用了50%的CPU，并且有8个核心。然后，对于绑定CPU的应用程序来说，要实现高效的多线程处理，就必须具备

8* (0.5) =4

如果您有4个线程，那么您的所有核心将有效地处理。这在一些支持超线程的公猪中发生了变化。