多线程的一般答案是通过堆栈(后进先出或先入先出)来取消递归实现。当实现这样的算法时,线程的数量是算法的固定参数(例如核心的数量)。
为了实现它,当测试条件结束递归时,语言调用堆栈被替换为存储最后上下文作为检查点的堆栈。在你的情况下是k=0
or coeff
值与目标总和匹配。
去递归之后,第一个实现是运行多个线程来消耗堆栈,但堆栈访问成为争用点,因为它可能需要同步。
更好的可扩展解决方案是为每个线程专用一个堆栈,但需要在堆栈中初始生成上下文。
我提出了一种混合方法,其中第一个线程递归地工作有限数量的线程k
作为最大递归深度:示例中的小数据集为 2,但如果较大,我建议使用 3。然后,第一部分将生成的中间上下文委托给一个线程池,该线程池将处理剩余的内容k
具有非递归实现。该代码不是基于您使用的复杂算法,而是基于相当“基本”的实现:
import java.util.Arrays;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class MixedParallel
{
// pre-requisite: sorted values !!
private static final int[] data = new int[] { 5, 10, 20, 25, 40, 50 };
// Context to store intermediate computation or a solution
static class Context {
int k;
int sum;
int[] coeff;
Context(int k, int sum, int[] coeff) {
this.k = k;
this.sum = sum;
this.coeff = coeff;
}
}
// Thread pool for parallel execution
private static ExecutorService executor;
// Queue to collect solutions
private static Queue<Context> solutions;
static {
final int numberOfThreads = 2;
executor =
new ThreadPoolExecutor(numberOfThreads, numberOfThreads, 1000, TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>());
// concurrent because of multi-threaded insertions
solutions = new ConcurrentLinkedQueue<Context>();
}
public static void main(String[] args)
{
int target_sum = 100;
// result vector, init to 0
int[] coeff = new int[data.length];
Arrays.fill(coeff, 0);
mixedPartialSum(data.length - 1, target_sum, coeff);
executor.shutdown();
// System.out.println("Over. Dumping results");
while(!solutions.isEmpty()) {
Context s = solutions.poll();
printResult(s.coeff);
}
}
private static void printResult(int[] coeff) {
StringBuffer sb = new StringBuffer();
for (int i = coeff.length - 1; i >= 0; i--) {
if (coeff[i] > 0) {
sb.append(data[i]).append(" * ").append(coeff[i]).append(" ");
}
}
System.out.println(sb.append("from ").append(Thread.currentThread()));
}
private static void mixedPartialSum(int k, int sum, int[] coeff) {
int x_k = data[k];
for (int c = sum / x_k; c >= 0; c--) {
coeff[k] = c;
int[] newcoeff = Arrays.copyOf(coeff, coeff.length);
if (c * x_k == sum) {
//printResult(newcoeff);
solutions.add(new Context(0, 0, newcoeff));
continue;
} else if (k > 0) {
if (data.length - k < 2) {
mixedPartialSum(k - 1, sum - c * x_k, newcoeff);
// for loop on "c" goes on with previous coeff content
} else {
// no longer recursive. delegate to thread pool
executor.submit(new ComputePartialSum(new Context(k - 1, sum - c * x_k, newcoeff)));
}
}
}
}
static class ComputePartialSum implements Callable<Void> {
// queue with contexts to process
private Queue<Context> contexts;
ComputePartialSum(Context request) {
contexts = new ArrayDeque<Context>();
contexts.add(request);
}
public Void call() {
while(!contexts.isEmpty()) {
Context current = contexts.poll();
int x_k = data[current.k];
for (int c = current.sum / x_k; c >= 0; c--) {
current.coeff[current.k] = c;
int[] newcoeff = Arrays.copyOf(current.coeff, current.coeff.length);
if (c * x_k == current.sum) {
//printResult(newcoeff);
solutions.add(new Context(0, 0, newcoeff));
continue;
} else if (current.k > 0) {
contexts.add(new Context(current.k - 1, current.sum - c * x_k, newcoeff));
}
}
}
return null;
}
}
}
您可以检查哪个线程找到了输出结果,并检查所有涉及的线程:递归模式下的主线程和上下文堆栈模式下池中的两个线程。
现在这个实现是可扩展的data.length
高:
- 最大递归深度限制在低级别的主线程
- 池中的每个线程都使用自己的上下文堆栈,不会与其他线程争用
- 现在要调整的参数是
numberOfThreads
and maxRecursionDepth
所以答案是肯定的,你的算法可以并行化。这是基于您的代码的完全递归实现:
import java.awt.Point;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class OriginalParallel
{
static final int numberOfThreads = 2;
static final int maxRecursionDepth = 3;
public static void main(String[] args)
{
int target_sum = 100;
int[] data = new int[] { 50, 40, 25, 20, 10, 5 };
List T = tableGeneator(target_sum, data);
int[] coeff = new int[data.length];
Arrays.fill(coeff, 0);
RecursivelyListAllThatWork(data.length, target_sum, T, coeff, data);
executor.shutdown();
}
private static void printResult(int[] coeff, int[] data) {
StringBuffer sb = new StringBuffer();
for (int i = coeff.length - 1; i >= 0; i--) {
if (coeff[i] > 0) {
sb.append(data[i]).append(" * ").append(coeff[i]).append(" ");
}
}
System.out.println(sb.append("from ").append(Thread.currentThread()));
}
// Thread pool for parallel execution
private static ExecutorService executor;
static {
executor =
new ThreadPoolExecutor(numberOfThreads, numberOfThreads, 1000, TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>());
}
private static void RecursivelyListAllThatWork(int k, int sum, List T, int[] coeff, int[] data) {
if (k == 0) {
printResult(coeff, data);
return;
}
Integer x_k = data[k-1];
// Recursive step: Try all coefficients, but only if they work.
for (int c = 0; c <= sum/x_k; c++) { //the c variable caps the percent
if (T.contains(new Point((sum - c * x_k), (k-1)))) {
// mark the coefficient of x_k to be c
coeff[k-1] = c;
if (data.length - k != maxRecursionDepth) {
RecursivelyListAllThatWork((k - 1), (sum - c * x_k), T, coeff, data);
} else {
// delegate to thread pool when reaching depth 3
int[] newcoeff = Arrays.copyOf(coeff, coeff.length);
executor.submit(new RecursiveThread(k - 1, sum - c * x_k, T, newcoeff, data));
}
// unmark the coefficient of x_k
coeff[k-1] = 0;
}
}
}
static class RecursiveThread implements Callable<Void> {
int k;
int sum;
int[] coeff;
int[] data;
List T;
RecursiveThread(int k, int sum, List T, int[] coeff, int[] data) {
this.k = k;
this.sum = sum;
this.T = T;
this.coeff = coeff;
this.data = data;
System.out.println("New job for k=" + k);
}
public Void call() {
RecursivelyListAllThatWork(k, sum, T, coeff, data);
return null;
}
}
public static List tableGeneator(int target_sum, int[] data) {
List T = new ArrayList();
T.add(new Point(0, 0));
float max_percent = 1;
int R = (int) (target_sum * max_percent * data.length);
for (int i = 0; i < data.length; i++) {
for (int s = -R; s < R + 1; s++) {
int max_value = (int) Math.abs((target_sum * max_percent) / data[i]);
for (int c = 0; c < max_value + 1; c++) {
if (T.contains(new Point(s - c * data[i], i))) {
Point p = new Point(s, i + 1);
if (!T.contains(p)) {
T.add(p);
}
}
}
}
}
return T;
}
}