一般问题
众所周知，Phaser 可用于同步所有任务的开始时间，如 JavaDocs 中提到的。和 this blog作者:尼克拉斯·施利姆。
Niklas 绘制了一幅非常容易理解的同步图像:
|Phaser |Phaser |Phaser |
Task 1 | ------> | ------> | ------> | ...
Task 2 | ------> | ------> | ------> | ...
...
现在假设有一个任务层次结构:
|Phaser |Phaser |Phaser |Phaser |Phaser |Phaser |Phaser | ...
Master | | | ------> | | | ------> | | ...
Task 1.1 | ----------------> | | ----------------> | | ----------> ...
Task 1.2 | ----------------> | | ----------------> | | ----------> ...
... | | | | | | | | ...
Task 2.1 | ------> | | | ------> | | | ------> | ...
Task 2.2 | ------> | | | ------> | | | ------> | ...
... | | | | | | | | ...
Task 3.1 | | ------> | | | ------> | | | ...
Task 3.2 | | ------> | | | ------> | | | ...
... | | | | | | | | ...
所以依赖树是这样的:
Master
/-----------/ \-----------\
| Task 2
Task 1 |
| Task 3
\-----------\ /-----------/
Master'
在一般情况下，需要解决一棵依赖树(假设在游戏管道中，有些是 AI/游戏逻辑/渲染任务)。幸运的是，有一个“大”同步点，并且树是固定的(但参与方的数量不是固定的)。使用多个移相器来求解是微不足道的。但是否可以只使用一个移相器？
一个特殊情况
具体来说，我编写了一个程序来解决以下问题。
|phasers[0]|phasers[1]|phasers[2]|phasers[0]|phasers[1]|phasers[2]| ...
Task 1 | -------> | | | -------> | | | ...
Task 2 | -------> | | | -------> | | | ...
Task 3 | | | -------> | | | -------> | ...
Task 4 | | -------> | | | -------> | | ...
代码在这里:
public class VolatileTester {
private int a = 0, b = 0; // change to volatile here
private int c = 0;
private final int TEST_COUNT = 100_000;
private int[] testResult = new int[TEST_COUNT];
private static void printResult(int[] result) {
final Map<Integer, Integer> countMap = new HashMap<>();
for (final int n : result) {
countMap.put(n, countMap.getOrDefault(n, 0) + 1);
}
countMap.forEach((n, count) -> {
System.out.format("%d -> %d%n", n, count);
});
}
private void runTask1() {
a = 5;
b = 10;
}
private void runTask2() {
if (b == 10) {
if (a == 5) {
c = 1;
} else {
c = 2;
}
} else {
if (a == 5) {
c = 3;
} else {
c = 4;
}
}
}
private void runTask3() {
// "reset task"
a = 0;
b = 0;
c = 0;
}
private static class PhaserRunner implements Runnable {
private final Phaser loopStartPhaser;
private final Phaser loopEndPhaser;
private final Runnable runnable;
public PhaserRunner(Phaser loopStartPhaser, Phaser loopEndPhaser, Runnable runnable) {
this.loopStartPhaser = loopStartPhaser;
this.loopEndPhaser = loopEndPhaser;
this.runnable = runnable;
}
@Override
public void run() {
while (loopStartPhaser.arriveAndAwaitAdvance() >= 0) {
runnable.run();
loopEndPhaser.arrive();
}
}
}
void runTest() throws InterruptedException {
final Phaser[] phasers = new Phaser[]{new Phaser(3), new Phaser(3), new Phaser(2)};
final Thread[] threads = new Thread[]{
// build tree of dependencies here
new Thread(new PhaserRunner(phasers[0], phasers[1], this::runTask1)),
new Thread(new PhaserRunner(phasers[0], phasers[1], this::runTask2)),
new Thread(new PhaserRunner(phasers[2], phasers[0], this::runTask3))
};
try {
for (Thread thread : threads) {
thread.start();
}
phasers[0].arrive(); // phaser of last round
for (int i = 0; i < TEST_COUNT; i++) {
phasers[1].arriveAndAwaitAdvance();
// Task4 here
testResult[i] = c;
phasers[2].arrive();
}
} finally {
for (Phaser phaser : phasers) {
phaser.forceTermination();
}
}
for (Thread thread : threads) {
thread.join();
}
printResult(testResult);
}
}
您可以看到使用了多个Phaser。是保留多个移相器(如上)更好，还是只使用一个大移相器？或者Java中还有其他推荐的同步方法吗？
最佳答案
应用程序中的所有任务都以逐步方式进行，这意味着单个移相器就足够了。要求是任务可以循环地跳过阶段，例如，对于每三个阶段，给定任务应该运行两个阶段，然后跳过一个阶段(一个阶段空闲)。综上所述，
- 任务应执行
arriveAndAwaitAdvance()在每一步工作之前。 - 任务应该只调用
arriveAndAwaitAdvance()跳过一个阶段。
为此，每个任务都可以使用一个 boolean 数组，在示例中称为 enabled ，指定是否在给定的阶段号启用。
通过使用模代数(enabled[phase % enabled.length])我们可以定义循环模式。例如，要指定任务应运行三个周期中的一个，我们声明 enabled如new boolean[]{true, false, false} .
请记住，无论任务是否执行任何实际工作，任务都必须推进该阶段。
我相应地修复了您的示例:
import java.util.concurrent.*;
import java.util.*;
public class VolatileTester {
private int a = 0, b = 0; // change to volatile here
private int c = 0;
private final int TEST_COUNT = 100;
private int[] testResult = new int[TEST_COUNT];
private static void printResult(int[] result) {
final Map<Integer, Integer> countMap = new HashMap<>();
for (final int n : result) {
countMap.put(n, countMap.getOrDefault(n, 0) + 1);
}
countMap.forEach((n, count) -> {
System.out.format("%d -> %d%n", n, count);
});
}
private void runTask1() {
a = 5;
b = 10;
}
private void runTask2() {
if (b == 10) {
if (a == 5) {
c = 1;
} else {
c = 2;
}
} else {
if (a == 5) {
c = 3;
} else {
c = 4;
}
}
}
private void runTask3() {
// "reset task"
a = 0;
b = 0;
c = 0;
}
private static class PhaserRunner implements Runnable {
private final Phaser phaser;
private final Runnable runnable;
private boolean[] enabled;
public PhaserRunner(Phaser phaser, boolean[] enabled, Runnable runnable) {
this.phaser = phaser;
this.runnable = runnable;
this.enabled = enabled;
}
@Override
public void run() {
int phase;
for (;;) {
phase = phaser.arriveAndAwaitAdvance();
if (phase < 0) {
break;
} else if (enabled[phase % enabled.length]) {
System.out.println("I'm running: " + Thread.currentThread());
runnable.run();
}
}
}
}
public void runTest() throws InterruptedException {
final Phaser phaser = new Phaser(4);
final Thread[] threads = new Thread[]{
// build tree of dependencies here
new Thread(new PhaserRunner(phaser, new boolean[]{true, false, false}, this::runTask1), "T1"),
new Thread(new PhaserRunner(phaser, new boolean[]{false, false, true}, this::runTask2), "T2"),
new Thread(new PhaserRunner(phaser, new boolean[]{false, true, false}, this::runTask3), "T3")
};
try {
for (Thread thread : threads) {
thread.start();
}
for (int i = 0; i < TEST_COUNT; i++) {
testResult[i] = c;
phaser.arriveAndAwaitAdvance();
}
} finally {
phaser.forceTermination();
}
for (Thread thread : threads) {
thread.join();
}
printResult(testResult);
}
public static void main(String[]args) throws Exception {
new VolatileTester().runTest();
}
}
关于java - 移相器同步用法，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/37157363/