一种减少多线程Java应用的工作队列中的竞争和开销的方法

原文地址:http://select.yeeyan.org/view/213582/202031

 

许多的服务器应用，比如说Web服务器、应用服务器、数据库服务器、文件服务器和邮件服务器等，这些应用都维 护着一些工作队列（worker queue）和线程池来处理大量从远程的来源处到达的短期任务。一般情况下，“工作队列”存放着所有需要执行的短期任务，线程池中的线程从工作队列中检索 任务并完成任务。由于多个线程在工作队列上操作，因此在工作队列中添加和删除任务需要做同步，而这就在工作队列中引入了竞争。本文解释了传统的方法（线程 池使用一个共同的队列）涉及的竞争，并通过为每个线程维护一个队列来帮助减少竞争。本文还介绍了一种工作任务窃取技术，在有效地利用多核系统中的CPU方 面，这一技术很重要。

前言

---

许多的服务器应用，比如说Web服务器、应用服务器、数据库服务器、文件服务器和邮件服务器等，这些应用都维护着一些工作队列（worker queue）和线程池来处理大量从远程的来源处到达的短期任务。一般情况下，“工作队列”存放着所有需要执行的短期任务，线程池中的线程从工作队列中检索 任务并完成任务。

由于多个线程在工作队列上操作，因此在工作队列中添加和删除任务需要做同步，而这就在工作队列中引入了竞争。本文解释了传统的方法（线程池使用一 个共同的队列）涉及的竞争，并通过为每个线程维护一个队列来帮助减少竞争。本文还介绍了一种工作任务窃取技术，在有效地利用多核系统中的CPU方面，这一 技术很重要。

注：本文中描述的例子的源代码可在这里下载： workerqueue.zip

共用的工作队列：传统的方法

---

目前，大部分的服务器应用都使用一个共用的工作队列和一个线程池来利用底层的硬件提供的并发性。如图1所示，服务器应用使用一个共用的工作队列来 存放从远端的来源处到达的任务。一个线程池在工作进程上操作，从工作队列中检索任务并执行任务直至完成。如果工作队列中没有任务的话，线程就阻塞在队列 上。

这一使用共用工作队列的方法解决了一些由早先的方法带来的问题，比如说每个任务创建一个线程，这种方法导致了大量线程的产生。然而，在任务数量非 常大且任务时间很短时，这一共用工作队列方法引来了一个瓶颈。当应用有数量庞大的短持续期的、独立的任务时，单后台线程方法也有缺陷。

.图1. 共用的工作队列

清单1展示了如何只需要几行代码就可以创建一个共用的工作队列。

清单1. 创建一个共用的工作队列

* 定义共用的工作队列和线程池来执行来自远端来源处的任务
*/
public class SimpleWorkQueue {
    private final PoolWorker[] threads;
    private final BlockingDeque queue;

    public SimpleWorkQueue(int nThreads)
    {
        queue = new LinkedBlockingDeque();
        threads = new PoolWorker[nThreads];
        for (int i=0; i<="" span="">
            threads[i] = new PoolWorker();
            threads[i].start();
        }
    }

    /*
     * 工作线程执行远程任务
     */
    private class PoolWorker extends Thread {              
        /*
         * 方法从工作队列中检索任务并开始执行任务
         * 如果队列中没有任务的话线程将等待
         */
        public void run() {
            while (!stopNow) {
                try {
           Runnable r = (Runnable) queue.takeLast();
           r.run();
                } catch ( java.lang.Throwable  e) { }
            }
        }
    }
}

如清单1所示，SimpleWorkQueue初始化了一个双端队列（dequeue）并在启动时开始执行固定数量的线程。然后每个线程在循环中 执行queue.takeLast()，该方法从工作队列中检索一个任务（如果有任务存在的话），或者等待新任务的到达（如果发现队列为空的话）。一但有 任务被检索到，每个线程接着调用任务的run方法r.run()。

每个线程一个工作队列

---

上面的方法非常简单，并且改进了为每个进来的任务都创建线程的这种传统做法的性能。然而，正如图2所示的那样，该方法带来了竞争。当多个线程使用单个工作队列来获取它们的任务时，竞争就产生了。线程（核）的数目越多时，竞争就越加的恶化。

图2. 共用工作队列中的竞争

目前，随着越来越多的多核处理器的问世，对于软件应用来说，有效利用底层的多核就成为了一项挑战。（例如， IBM的Power7、Oracle的UltraSPARC和Intel's Nehalem都是具有运行多线程能力的多核处理器）

有多种解决方案可用来克服共用工作队列方法中的竞争。

1. 使用不加锁的数据结构

2. 使用多锁的并发数据结构

3. 维护多个队列以隔离竞争

在本文中，我们介绍如何维护多个队列——每个线程一个队列（queue-per-thread）的方法——以此来隔离竞争，如图3所示。

图3. 每个线程一个队列式的队列

在这一方法中，每个线程都有自己的工作队列，并且只能从自己的队列而不能从任何的其他队列中检索任务。该方法隔离了检索任务时的竞争，因为不存在其他要和它争夺任务的线程。这一做法保证了如果工作队列中有任务存在的话，线程就不会进入睡眠状态，这样就有效地利用到了多核。

清单2展示了如何很容易地从共用工作队列方法迁移到每个线程一个队列方法上，只需对清单1展示的代码做几处修改就可以了。在清单2中，构造函数在 启动时初始化了多个队列（等于线程的数目），每个线程都维护一个名为thread_id的ID。接着，thread_id被用来隔离竞争，其帮助每个线程 从自己的队列中检索任务。

清单2. 创建每个线程一个队列式的队列

/*
/* 修改成多个队列的初始化*/
for (int i=0; i();
}
...
......
/* 任务检索的修改 */
r = (Runnable) queue[thread_id].takeLast();

  

使用了工作窃取手段的每个线程一个队列式的队列

---

尽管每个线程一个队列这种方法极大地减少了竞争，但它并没有保证底层的多核在所有时候都能够被有效利用。例如，如果有一两个队列比其他队列先变空 了的话会有什么事情发生呢？这是一种常见的情况，在这种情况下，只有少数的线程在执行任务，而其他的线程（队列已空的线程）则在等待新任务的到来。这种情 况是可能发生的，理由如下：

1. 调度算法的不可预测性

2. 传入任务的不可预测性（短持续时间的和长持续时间的）

这一问题的一种解决方案是工作窃取（work stealing）。

当一个线程发现自己的队列变空时，工作窃取手段让该线程从另一该队列中窃取工作，这种做法确保了所有的线程（因此相应的，多个核）每时每刻都是忙 碌的。图4展示了这样的一种场景，线程2从线程1的队列中窃取了一个工作任务，因为它自己的队列已空。工作窃取可使用一般的队列来实现，不过使用一个双端 队列则会极大的减少窃取工作过程所涉及的竞争：

1. 只有工作线程才能访问它自己的双端队列的头端，因此双端队列的头端永远也不会存在竞争。

2. 双端队列的尾端只有在线程已经运行完所有的工作时才会访问到，因此任何线程的双端队列的尾端也都很少有竞争出现。

图4. 工作窃取

清单3说明了如何从其他的队列中窃取工作任务，只需要对每个线程一个队列方法做几处修改就可以了。正如所展示的那样，每个线程都调用 pollLast()而不是takeLast()。这是必需的，因为线程不应该因为队列中没有任务而阻塞在队列上。一旦线程发现它自己的队列变空了的话， 它就通过调用其他线程队列的pollFirst()来从其他队列中窃取工作任务。

清单3. 实现工作窃取

/* 如果当前队列中没有任务的话不阻塞 */
r = (Runnable) queue[thread_id].pollLast();

if(null == r) {
    /* 如果当前队列中没有任务的话，从另一个线程的队列中窃取一个 */
    r = stealWork(thread_id);
}

/*
* 从另一个队列中窃取工作任务的方法
*/
Runnable stealWork(int index) {
        for (int i=0; i<="" span="">
            if(i != index) {
                Object o = queue[i].pollFirst();
                if(o!=null) {
                    return (Runnable) o;
                }
            }
        }
        return null;
}

  <="" pre="">

构建基准（benchmark）

---

为了说明这三个方法，我们为本文中提到的三个方法定制了一个小型的测试方案并研究其行为。这一测试的基本工作是创建许多的10x10矩阵乘法数量的任务并使用三个方法来执行它们。

注：本文中描述的例子的代码可从这里下载： workerqueue.zip

该测试定义了下面的类：

<="" pre=""> 1. MainClass：该类启动、开始和协调基准的各个元素。

<="" pre="">2. WorkAssignerThread<="" pre="">：该线程类创建大量的10x10矩阵乘法数量的任务并队列化它们。

<="" pre="">3. Task：该类定义一个10x10的矩阵乘法。

<="" pre="">4. WorkQueue：该接口定义了一组任何工作队列都必须要实现的方法。

<="" pre="">5. WorkerQuerueFactory：该工厂类基于队列类型返回workQueue对象。

<="" pre="">6. SimpleWorkQueue：该类定义了一个简单的工作队列并启动一组线程。这描述的是本文中提到的第一种队列类型（共用的工作队列）。

<="" pre="">7. MultiWorkQueue：该类通过定义多个工作队列（每个线程一个）来隔离竞争，这描述的是本文中提到的第二种队列类型。

<="" pre="">8. WorkStealingQueue：该类通过定义多个队列来隔离竞争，并在其发现其线程的队列为空时窃取工作任务。这描述的是本文中提到的第三种队列类型。

<="" pre="">

该测试可通过指明队列类型、线程数目和任务数目来执行，如清单4所示。清单4还说明了如何使用第一种队列类型（共用的工作队列）、线程数为10和任务数为10000来调用测试。

清单4. 执行测试

java MainClass  

/* 例子：*/

<="" pre="">java MainClass 1 10 10000

<="" pre="">

实验结果

---

我们在不同的架构中评估了性能，结构都非常的乐观。最初的时候，我们在AMD Operon盒装处理器上进行性能评估，其有八个内核处理器，运行的是Linux，我们发现队列类型3的性能比队列类型1的提高了12%到18.4%，这取决于负载情况，如图5所示。

 图5. 队列类型1和队列类型3的性能比较，在带有八个内核处理器的Linux AMD Opteron系统上运行。[免责声明：这不是对任何产品的比较或是宣称任何产品的性能，这仅是本文所提出的技术优势的展示用例，纯属作者的观点。]

我们还在有着四个双核处理器（Power 4）的Linux power系统上进行了性能评估，我们发现在相同负载下，性能提升了12%到16%，如图6所示。

图6. 队列类型1和队列类型3之间的性能比较，运行在有着四个双核Power处理器的Linux系统上。[免责声明：这不是对任何产品的比较或是宣称任何产品的性能，这仅是本文所提出的技术优势的展示用例，纯属作者的观点。]

正如图5和图6所展示的那样，我们从10万到50万改变任务的数目，并以秒为单位来衡量性能。实验的结果清楚地证明，队列类型1产生了数量巨多的竞争，这些竞争可通过创建多队列和窃取工作任务手段来消除。

小结

---

本文说明了共用工作队列涉及的竞争，然后通过为每个线程创建一个队列来隔离竞争。文章还通过一个简单的测试基准来说明了为什么工作窃取手段很重要，以及这一做法如何提高了应用的整体性能。

原文地址:http://select.yeeyan.org/view/213582/202031