3

我正在编写使用 Phaser 来知道何时完成工作的多线程应用程序。问题是在 ExecutorCompletionService 中甚至可以有 100k 的线程在一个队列中，但在 Phaser 中的最大到达方数是 65535。当有 65536 方到达时我该怎么办？

我的示例代码：

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class Main {
public static void main(String[] args) throws Exception {
    ExecutorService ec = Executors.newFixedThreadPool(10);
    ExecutorCompletionService<List<String>> ecs = new ExecutorCompletionService<List<String>>(
            ec);
    Phaser phaser = new Phaser();

    // register first node/thread
    ecs.submit(new SimpleParser("startfile.txt"));
    phaser.register();

    Future<List<String>> future;
    do {
        future = ecs.poll();
        if(future!=null && future.get() != null) {
            addParties(phaser, future.get(), ecs);
            phaser.arriveAndDeregister();
        }

        if (phaser.isTerminated()) {
            ec.shutdown();
        }
    } while (!ec.isShutdown() && !phaser.isTerminated());
}

public static void addParties(Phaser p, List<String> filenames,
        ExecutorCompletionService<List<String>> ecs) {
    for (int i = 0; i < filenames.size(); i++) {
        ecs.submit(new SimpleParser(filenames.get(i)));
        //PROBLEM = What to do when Phaser has 65535+ unarrived parties
        p.register();
    }
}

static class SimpleParser implements Callable<List<String>> {

    String fileName;

    public SimpleParser(String fileName) {
        this.fileName = fileName;
    }

    @Override
    public List<String> call() throws Exception {
        return parseFile();
    }

    private List<String> parseFile() {
        return new ArrayList<String>(Arrays.asList(new String[] {
                "somefilename1.txt", "somefilename2.txt" }));
    }

}
}

问题出在 addParties() 方法中。单线程（SimpleParser）可以返回即100个新文件名，将有100个新线程提交给ExecutorCompletionService，100个新方在Phaser中注册。我试图使用这样的东西：

if(p.getUnarrivedParties() == 65535)
            p = new Phaser(p);

并创建一个相位器链，但它没有帮助，因为 p.getUnarrivedParties() 返回 0，但我无法注册下一个参与方...

    System.out.println(p.getUnarrivedParties());
        if(p.getUnarrivedParties() == 65535) {
            p = new Phaser(p);
            System.out.println(p.getUnarrivedParties());
        }
        p.register();

印刷：

65535

0

并抛出 IllegalStateException

那么如何创建与这个旧的Phaser 连接的新Phaser？

//编辑

谢谢@bowmore。我还有两个问题。

让我们看一下这个例子：

import java.util.concurrent.Phaser;

public class Test2 {
    public static void main(String[] args) {
        Phaser parent = new Phaser();
        Phaser child1 = new Phaser(parent);
        Phaser child2 = new Phaser(parent);
        child1.register();
        child2.register();

        System.out.println("Parent: "+parent.isTerminated());
        System.out.println("Child1: "+child1.isTerminated());
        System.out.println("Child2: "+child1.isTerminated()+"\n");

        child1.arriveAndDeregister();
        System.out.println("Parent: "+parent.isTerminated());
        System.out.println("Child1: "+child1.isTerminated());
        System.out.println("Child2: "+child2.isTerminated()+"\n");

        child2.arriveAndDeregister();
        System.out.println("Parent: "+parent.isTerminated());
        System.out.println("Child1: "+child1.isTerminated());
        System.out.println("Child2: "+child2.isTerminated()+"\n");
    }
}

它打印：

Parent: false
Child1: false
Child2: false

Parent: false
Child1: false
Child2: false

Parent: true
Child1: true
Child2: true

为什么在 child1.arriveAndDeregister(); 之后 child1 没有终止，如何检查它是否真的终止？

第二个问题。我在达到 65535 方后询问是否要创建新的 Phaser，因为我认为创建数千个新对象是没有用的 - 你认为这样做不会有内存问题，或者它甚至可以提高性能吗？

4

1 回答 1

3

可以在新创建的原始子进程上注册，而不是在现有的Phaser新进程中注册。只需将父级提供给子级的构造函数即可Phaser创建子级。PhaserPhaser

public static void addParties(Phaser p, List<String> filenames,
                              ExecutorCompletionService<List<String>> ecs) {
    Phaser newPhaser = new Phaser(p);
    for (int i = 0; i < filenames.size(); i++) {
        ecs.submit(new SimpleParser(filenames.get(i)));
        newPhaser.register();
    }
}

如果您只想在达到某个阈值时创建子 Phaser，您可以检查已注册方的数量，而不是未到达者的数量：

public static void addParties(Phaser p, List<String> filenames, ExecutorCompletionService<List<String>> ecs) {
    Phaser toRegister = p.getRegisteredParties() > THRESHOLD ? new Phaser(p) : p;
    for (int i = 0; i < filenames.size(); i++) {
        ecs.submit(new SimpleParser(filenames.get(i)));
        //PROBLEM = What to do when Phaser has 65535+ unarrived parties
        toRegister.register();
    }
    System.out.println(p.getRegisteredParties());
}

编辑 ：

跟进问题 1： ChildPhaser与 root 共享终止状态Phaser，这里是实现isTerminated()

public boolean isTerminated() {
    return root.state < 0L;
}

跟进问题 2：父 Phasers 实际上并没有保留对其子 Phasers 的引用。一旦不再引用子移相器，它就有资格进行垃圾收集。您最好遵循 javadoc 中的建议：

TASKS_PER_PHASER 的最佳值主要取决于预期的同步速率。对于极小的每阶段任务主体（因此高速率），低至四的值可能适合，对于非常大的任务主体，可能适合高达数百。

分层的主要原因是减少繁重的同步争用，因此如果您有轻量级任务，则每个移相器的任务越少越好。配置不同的设置来调整这些东西永远不会有坏处。

于 2012-12-30T20:46:11.103 回答