我正在对大量数据进行快速排序，为了好玩，我试图将其并行化以加快排序速度。但是，在当前形式中，由于同步阻塞点，多线程版本比单线程版本慢。
每次我生成一个线程时，我都会在一个 int 上获得一个锁并递增它，每次线程完成时我都会再次获得一个锁和递减，此外还要检查是否有任何线程仍在运行（int > 0）。如果没有，我会唤醒我的主线程并处理已排序的数据。
我确信有更好的方法来做到这一点。虽然不确定它是什么。非常感谢您的帮助。
编辑：我想我没有提供足够的信息。
这是一个八核 Opteron 上的 Java 代码。我不能切换语言。
我正在排序的数量适合内存，并且在调用快速排序时它已经存在于内存中，因此没有理由将其写入磁盘只是为了将其读回内存。
“获得锁”是指在整数上有一个同步块。
在不了解更多有关实施的情况下，这里是我的建议和/或评论：
限制可以在任何给定时间运行的线程数。Pergaps 8 或 10（也许给调度程序更多的回旋余地，尽管最好为每个核心/硬件线程放置一个）。如果亲和力不支持它，那么在 CPU 绑定问题上运行更多线程以实现“吞吐量”实际上没有任何意义。
不要在树叶附近穿线！！！只在较大的树枝上穿线。产生一个线程来对数量相对较少的项目进行排序是没有意义的，在这个级别有很多很多小分支！线程在这里会增加更多的相对开销。（这类似于切换到叶子的“简单排序”）。
确保每个线程都可以独立工作-在工作期间不应踩到另一个线程->没有锁，只需等待 join。分而治之。
可能会考虑执行“广度优先”的方法来生成线程。
考虑一下快速排序上的合并排序（我偏向于合并排序:-) 请记住，有许多不同类型的合并排序，包括自下而上。
编辑
我把这个简单的演示放在一起。在我的 Intel Core2 Duo @ 2Ghz 上，我可以让它在大约 2/3 到 3/4 的时间内运行，这绝对是一些改进:)（设置：DATA_SIZE = 3000000，MAX_THREADS = 4，MIN_PARALLEL = 1000）。这是从 Wikipedia 中提取的基本就地快速排序代码，它没有利用任何其他基本优化。
它确定一个新线程是否可以/应该启动的方法也是非常原始的——如果没有新线程可用，它就会继续前进（因为，你知道，为什么要等待？）
这段代码也应该（希望）与线程一起扇出广度。对于数据局部性而言，这可能比保持深度方面效率低，但如果我的头脑，该模型似乎足够简单。
执行器服务还用于简化设计并能够重用相同的线程（相对于产生新线程）。在执行程序开销开始显示之前，MIN_PARALLEL 可能会变得非常小（例如，大约 20）——最大线程数和 only-using-a-new-thread-if-possible 也可能会控制这一点。
qsort 平均秒数：0.6290541056 pqsort 平均秒数：0.4513915392
我绝对不保证这段代码的有用性或正确性，但它在这里“似乎有效”。注意 ThreadPoolExecutor 旁边的警告，因为它清楚地表明我不完全确定发生了什么 :-)我相当肯定该设计在未充分利用线程方面存在一定缺陷。
package psq;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.*;
public class Main {
int[] genData (int len) {
Random r = new Random();
int[] newData = new int[len];
for (int i = 0; i < newData.length; i++) {
newData[i] = r.nextInt();
}
return newData;
}
boolean check (int[] arr) {
if (arr.length == 0) {
return true;
}
int lastValue = arr[0];
for (int i = 1; i < arr.length; i++) {
//System.out.println(arr[i]);
if (arr[i] < lastValue) {
return false;
}
lastValue = arr[i];
}
return true;
}
int partition (int[] arr, int left, int right, int pivotIndex) {
// pivotValue := array[pivotIndex]
int pivotValue = arr[pivotIndex];
{
// swap array[pivotIndex] and array[right] // Move pivot to end
int t = arr[pivotIndex];
arr[pivotIndex] = arr[right];
arr[right] = t;
}
// storeIndex := left
int storeIndex = left;
// for i from left to right - 1 // left ≤ i < right
for (int i = left; i < right; i++) {
//if array[i] ≤ pivotValue
if (arr[i] <= pivotValue) {
//swap array[i] and array[storeIndex]
//storeIndex := storeIndex + 1
int t = arr[i];
arr[i] = arr[storeIndex];
arr[storeIndex] = t;
storeIndex++;
}
}
{
// swap array[storeIndex] and array[right] // Move pivot to its final place
int t = arr[storeIndex];
arr[storeIndex] = arr[right];
arr[right] = t;
}
// return storeIndex
return storeIndex;
}
void quicksort (int[] arr, int left, int right) {
// if right > left
if (right > left) {
// select a pivot index //(e.g. pivotIndex := left + (right - left)/2)
int pivotIndex = left + (right - left) / 2;
// pivotNewIndex := partition(array, left, right, pivotIndex)
int pivotNewIndex = partition(arr, left, right, pivotIndex);
// quicksort(array, left, pivotNewIndex - 1)
// quicksort(array, pivotNewIndex + 1, right)
quicksort(arr, left, pivotNewIndex - 1);
quicksort(arr, pivotNewIndex + 1, right);
}
}
static int DATA_SIZE = 3000000;
static int MAX_THREADS = 4;
static int MIN_PARALLEL = 1000;
// NOTE THAT THE THREAD POOL EXECUTER USES A LINKEDBLOCKINGQUEUE
// That is, because it's possible to OVER SUBMIT with this code,
// even with the semaphores!
ThreadPoolExecutor tp = new ThreadPoolExecutor(
MAX_THREADS,
MAX_THREADS,
Long.MAX_VALUE,
TimeUnit.NANOSECONDS,
new LinkedBlockingQueue<Runnable>());
// if there are no semaphore available then then we just continue
// processing from the same thread and "deal with it"
Semaphore sem = new Semaphore(MAX_THREADS, false);
class QuickSortAction implements Runnable {
int[] arr;
int left;
int right;
public QuickSortAction (int[] arr, int left, int right) {
this.arr = arr;
this.left = left;
this.right = right;
}
public void run () {
try {
//System.out.println(">>[" + left + "|" + right + "]");
pquicksort(arr, left, right);
//System.out.println("<<[" + left + "|" + right + "]");
} catch (Exception ex) {
// I got nothing for this
throw new RuntimeException(ex);
}
}
}
// pquicksort
// threads will [hopefully] fan-out "breadth-wise"
// this is because it's likely that the 2nd executer (if needed)
// will be submitted prior to the 1st running and starting its own executors
// of course this behavior is not terribly well-define
void pquicksort (int[] arr, int left, int right) throws ExecutionException, InterruptedException {
if (right > left) {
// memory barrier -- pquicksort is called from different threads
synchronized (arr) {}
int pivotIndex = left + (right - left) / 2;
int pivotNewIndex = partition(arr, left, right, pivotIndex);
Future<?> f1 = null;
Future<?> f2 = null;
if ((pivotNewIndex - 1) - left > MIN_PARALLEL) {
if (sem.tryAcquire()) {
f1 = tp.submit(new QuickSortAction(arr, left, pivotNewIndex - 1));
} else {
pquicksort(arr, left, pivotNewIndex - 1);
}
} else {
quicksort(arr, left, pivotNewIndex - 1);
}
if (right - (pivotNewIndex + 1) > MIN_PARALLEL) {
if (sem.tryAcquire()) {
f2 = tp.submit(new QuickSortAction(arr, pivotNewIndex + 1, right));
} else {
pquicksort(arr, pivotNewIndex + 1, right);
}
} else {
quicksort(arr, pivotNewIndex + 1, right);
}
// join back up
if (f1 != null) {
f1.get();
sem.release();
}
if (f2 != null) {
f2.get();
sem.release();
}
}
}
long qsort_call (int[] origData) throws Exception {
int[] data = Arrays.copyOf(origData, origData.length);
long start = System.nanoTime();
quicksort(data, 0, data.length - 1);
long duration = System.nanoTime() - start;
if (!check(data)) {
throw new Exception("qsort not sorted!");
}
return duration;
}
long pqsort_call (int[] origData) throws Exception {
int[] data = Arrays.copyOf(origData, origData.length);
long start = System.nanoTime();
pquicksort(data, 0, data.length - 1);
long duration = System.nanoTime() - start;
if (!check(data)) {
throw new Exception("pqsort not sorted!");
}
return duration;
}
public Main () throws Exception {
long qsort_duration = 0;
long pqsort_duration = 0;
int ITERATIONS = 10;
for (int i = 0; i < ITERATIONS; i++) {
System.out.println("Iteration# " + i);
int[] data = genData(DATA_SIZE);
if ((i & 1) == 0) {
qsort_duration += qsort_call(data);
pqsort_duration += pqsort_call(data);
} else {
pqsort_duration += pqsort_call(data);
qsort_duration += qsort_call(data);
}
}
System.out.println("====");
System.out.println("qsort average seconds: " + (float)qsort_duration / (ITERATIONS * 1E9));
System.out.println("pqsort average seconds: " + (float)pqsort_duration / (ITERATIONS * 1E9));
}
public static void main(String[] args) throws Exception {
new Main();
}
}
YMMV。快乐编码。
（另外，我想知道这个 - 或类似的 - 在你的 8 核盒子上的代码公平。维基百科声称通过 CPU 数量线性加速是可能的:)
删除了导致轻微“堵塞”的期货的使用，并切换到单个最终等待信号量：减少无用的等待。现在只用 55% 的非线程时间运行 :-)
qsort 平均秒数：0.5999702528 pqsort 平均秒数：0.3346969088
(
package psq;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.*;
public class Main {
int[] genData (int len) {
Random r = new Random();
int[] newData = new int[len];
for (int i = 0; i < newData.length; i++) {
newData[i] = r.nextInt();
}
return newData;
}
boolean check (int[] arr) {
if (arr.length == 0) {
return true;
}
int lastValue = arr[0];
for (int i = 1; i < arr.length; i++) {
//System.out.println(arr[i]);
if (arr[i] < lastValue) {
return false;
}
lastValue = arr[i];
}
return true;
}
int partition (int[] arr, int left, int right, int pivotIndex) {
// pivotValue := array[pivotIndex]
int pivotValue = arr[pivotIndex];
{
// swap array[pivotIndex] and array[right] // Move pivot to end
int t = arr[pivotIndex];
arr[pivotIndex] = arr[right];
arr[right] = t;
}
// storeIndex := left
int storeIndex = left;
// for i from left to right - 1 // left ≤ i < right
for (int i = left; i < right; i++) {
//if array[i] ≤ pivotValue
if (arr[i] <= pivotValue) {
//swap array[i] and array[storeIndex]
//storeIndex := storeIndex + 1
int t = arr[i];
arr[i] = arr[storeIndex];
arr[storeIndex] = t;
storeIndex++;
}
}
{
// swap array[storeIndex] and array[right] // Move pivot to its final place
int t = arr[storeIndex];
arr[storeIndex] = arr[right];
arr[right] = t;
}
// return storeIndex
return storeIndex;
}
void quicksort (int[] arr, int left, int right) {
// if right > left
if (right > left) {
// select a pivot index //(e.g. pivotIndex := left + (right - left)/2)
int pivotIndex = left + (right - left) / 2;
// pivotNewIndex := partition(array, left, right, pivotIndex)
int pivotNewIndex = partition(arr, left, right, pivotIndex);
// quicksort(array, left, pivotNewIndex - 1)
// quicksort(array, pivotNewIndex + 1, right)
quicksort(arr, left, pivotNewIndex - 1);
quicksort(arr, pivotNewIndex + 1, right);
}
}
static int DATA_SIZE = 3000000;
static int MAX_EXTRA_THREADS = 7;
static int MIN_PARALLEL = 500;
// To get to reducePermits
@SuppressWarnings("serial")
class Semaphore2 extends Semaphore {
public Semaphore2(int permits, boolean fair) {
super(permits, fair);
}
public void removePermit() {
super.reducePermits(1);
}
}
class QuickSortAction implements Runnable {
final int[] arr;
final int left;
final int right;
final SortState ss;
public QuickSortAction (int[] arr, int left, int right, SortState ss) {
this.arr = arr;
this.left = left;
this.right = right;
this.ss = ss;
}
public void run () {
try {
//System.out.println(">>[" + left + "|" + right + "]");
pquicksort(arr, left, right, ss);
//System.out.println("<<[" + left + "|" + right + "]");
ss.limit.release();
ss.countdown.release();
} catch (Exception ex) {
// I got nothing for this
throw new RuntimeException(ex);
}
}
}
class SortState {
final public ThreadPoolExecutor pool = new ThreadPoolExecutor(
MAX_EXTRA_THREADS,
MAX_EXTRA_THREADS,
Long.MAX_VALUE,
TimeUnit.NANOSECONDS,
new LinkedBlockingQueue<Runnable>());
// actual limit: executor may actually still have "active" things to process
final public Semaphore limit = new Semaphore(MAX_EXTRA_THREADS, false);
final public Semaphore2 countdown = new Semaphore2(1, false);
}
void pquicksort (int[] arr) throws Exception {
SortState ss = new SortState();
pquicksort(arr, 0, arr.length - 1, ss);
ss.countdown.acquire();
}
// pquicksort
// threads "fork" if available.
void pquicksort (int[] arr, int left, int right, SortState ss) throws ExecutionException, InterruptedException {
if (right > left) {
// memory barrier -- pquicksort is called from different threads
// and those threads may be created because they are in an executor
synchronized (arr) {}
int pivotIndex = left + (right - left) / 2;
int pivotNewIndex = partition(arr, left, right, pivotIndex);
{
int newRight = pivotNewIndex - 1;
if (newRight - left > MIN_PARALLEL) {
if (ss.limit.tryAcquire()) {
ss.countdown.removePermit();
ss.pool.submit(new QuickSortAction(arr, left, newRight, ss));
} else {
pquicksort(arr, left, newRight, ss);
}
} else {
quicksort(arr, left, newRight);
}
}
{
int newLeft = pivotNewIndex + 1;
if (right - newLeft > MIN_PARALLEL) {
if (ss.limit.tryAcquire()) {
ss.countdown.removePermit();
ss.pool.submit(new QuickSortAction(arr, newLeft, right, ss));
} else {
pquicksort(arr, newLeft, right, ss);
}
} else {
quicksort(arr, newLeft, right);
}
}
}
}
long qsort_call (int[] origData) throws Exception {
int[] data = Arrays.copyOf(origData, origData.length);
long start = System.nanoTime();
quicksort(data, 0, data.length - 1);
long duration = System.nanoTime() - start;
if (!check(data)) {
throw new Exception("qsort not sorted!");
}
return duration;
}
long pqsort_call (int[] origData) throws Exception {
int[] data = Arrays.copyOf(origData, origData.length);
long start = System.nanoTime();
pquicksort(data);
long duration = System.nanoTime() - start;
if (!check(data)) {
throw new Exception("pqsort not sorted!");
}
return duration;
}
public Main () throws Exception {
long qsort_duration = 0;
long pqsort_duration = 0;
int ITERATIONS = 10;
for (int i = 0; i < ITERATIONS; i++) {
System.out.println("Iteration# " + i);
int[] data = genData(DATA_SIZE);
if ((i & 1) == 0) {
qsort_duration += qsort_call(data);
pqsort_duration += pqsort_call(data);
} else {
pqsort_duration += pqsort_call(data);
qsort_duration += qsort_call(data);
}
}
System.out.println("====");
System.out.println("qsort average seconds: " + (float)qsort_duration / (ITERATIONS * 1E9));
System.out.println("pqsort average seconds: " + (float)pqsort_duration / (ITERATIONS * 1E9));
}
public static void main(String[] args) throws Exception {
new Main();
}
}
“获得锁”是指在整数上有一个同步块。如果我理解正确的话：你锁定了你实际排序的每一个元素，这听起来会很慢！
听起来你产生了太多线程......你还没有告诉我们你实际产生了多少线程，但是如果你每个整数做一个线程，那么它几乎肯定会更慢（几乎可以肯定是轻描淡写）。您想要做的是产生 8 个线程，因为您有 8 个内核，并将您的数组“分区”为 8 个部分，您将分别对其进行快速排序，然后像在原始算法中那样连接。
以下是一些关于如何实现它的示例：多线程快速排序或合并排序
线程很昂贵。如果您没有大量数据要排序，请不要使用线程。或者您可以使用具有更好的并发设计的语言。例如，Erlang 有非常轻量级的线程，可用于排序。