在java中同步两个线程

Question

我的 java 程序中有两个线程，一个是主线程，另一个线程是在主线程中产生的线程 A。现在我希望主线程启动线程 A 并等待线程 A 在 run 方法中执行了它的部分代码，并且线程 A 应该挂起自己。然后主线程应该开始运行，运行几行代码，然后线程 A 应该从它停止的地方开始，反之亦然。这应该发生 n 次。 我正在尝试如下：

线程A类：

public class ThreadA implements Runnable {
    boolean suspended = false;
    boolean stopped = false;
    synchronized void stop() {
        stopped = true;
        suspended = false;
        notify();
    }
    synchronized void suspend() {
        suspended = true;
    }
    synchronized void resume() {
        suspended = false;
        notify();
    }
    void job() throws InterruptedException {
        for (int i = 0; i < 5; i++)
            synchronized (this) {
                System.out.println("performing job.");
                suspend();
                while (suspended) {
                    notify();
                    suspended = false;
                }
            }
    }
    @Override
    public void run() {
        try {
            job();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

主线程：

public class MainThread {
    public static void main(String[] args) throws InterruptedException {
        ThreadA a1=new ThreadA();
        Thread t1=new Thread(a1);
        synchronized (t1) {
            t1.start();
            for (int i = 0; i < 5; i++) {
                t1.wait();
                System.out.println("perform some action");
                a1.resume();
            }

        }
    }
}

预期输出：

performing job.
perform some action
performing job.
perform some action
performing job.
perform some action
performing job.
perform some action
performing job.
perform some action

实际输出：

performing job.
performing job.
performing job.
performing job.
performing job.
perform some action

我不知道为什么整个 for 循环会在线程 A 中执行，即使我在作业方法中发出了 notify() 信号。

Answer 1

这里有两个错误。

首先是您正在同步和通知不同的对象。试试这个修改过的 main，我将 synchronized (t1) 更改为 synchronized (a1) 并将 t1.wait() 更改为 a1.wait()。

public static void main(String[] args) throws InterruptedException {
    ThreadA a1=new ThreadA();
    Thread t1=new Thread(a1);

    synchronized (a1) {    // CHANGED FROM t1 to a1
        t1.start();
        for (int i = 0; i < 5; i++) {
            a1.wait();    // CHANGED FROM t1 to a1
            System.out.println("perform some action");
            a1.resume();
        }

    }
}

第二个错误是在 job() 方法中，它调用 notify() 而不是 wait()。这是一个固定版本：

void job() throws InterruptedException {
    for (int i = 0; i < 5; i++)
        synchronized (this) {
            System.out.println("performing job.");
            suspend();
            while (suspended) {
                notify();
                suspended = false;
                wait();    // ADDED
            }
        }
}

我的测试运行的输出是

performing job.
perform some action
performing job.
perform some action
performing job.
perform some action
performing job.
perform some action
performing job.
perform some action

Answer 2

这里是更简化的方式

public class TwoThread {

    public static void main(String[] args) throws InterruptedException {
        ThreadA a1 = new ThreadA();
        Thread t1 = new Thread(a1);

        synchronized (a1) {
            t1.start();
            for (int i = 0; i < 5; i++) {
                a1.wait();
                System.out.println("perform some action " + i);
                a1.notify();
            }

        }
    }
}

public class ThreadA implements Runnable {
    boolean suspended = false;
    boolean stopped = false;

    void job() throws InterruptedException {
        for (int i = 0; i < 5; i++)
            synchronized (this) {
                System.out.println("performing job. " + i);
                notify();
                wait();
            }
    }

    public void run() {
        try {
            job();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}

Answer 3

仅在两个线程之间进行通信：

1. 您不需要同步
2. 您不需要锁
3. 您不需要 CAS（比较和交换）。 （不强不弱）
4. 您不需要 setOpaque、setVolative 或 setRelease

您只需要 VarHandle 屏障（在 Java 中）。 Java VarHandle 唯一的要求是有序的内存访问

这是一篇相当不错的文章。 Synchronizing without Locks and Concurrent Data Structures

在我的日常工作中，我使用 Dekker 和 Peterson 算法的变体 用于 Web 请求的异步多线程处理、共享连接池和从云应用程序收集日志，对性能影响较小 与单线程无竞争性能相比。

有时，我必须使用 setOpaque 和 getOpaque， 使用 VarHandle.loadLoadFence() 和 VarHandle.storeStoreFence() 以确保有序的内存访问，这就是您所需要的。 在我看来，弱 CAS 是我会走得更远的地方， 就像我认为违反多核 CPU 架构的其他任何事情一样。

但是，除非你对实际情况有深入的了解 您正在使用的硬件和内存排序结构 在微指令级别使用，我建议你使用 标准的 Java 并发锁，因为它们是最好的 最适合通用解决方案。

要实现比传统 CAS 算法提高 10 倍的性能， 您需要对内存中的所有共享对象进行非常稳定的布局 并严格定义哪些线程可以读取 以及可以写入每个变量的顺序。 您将需要考虑对 CPU 缓存的副作用 所有内存加载和存储，然后得到这些 在特定平台上发挥您的优势 你的目标。你最终会得到 相当复杂的算法，但性能无与伦比。

您应该探索 LMAX Disruptor 库 因为它有实现许多这些概念的开源库 像环形缓冲区和单线程可以写入每个变量。 LMAX Disruptor User Guide 然而，我仍然认为这是保守的并发方法。 我目前的标准是拥有能够容忍赛车的算法 如果检测到赛车状况，则丢弃数据并重复处理。 我使用状态嵌入计数器、索引、标志和哈希， 检测线程冲突并选择将放弃的线程 并使用另一个内存结构进行操作。 然而，由于线程集中的内存结构和 优化的参考共享很少发生（例如百万分之一）

但是，如果您对 CPU 缓存操作有很好的了解 以及您可以获得 CPU 的任何专用平台指令 缓存以发挥您的优势并执行读取和写入， 以及在内核之间共享缓存线，作为副作用 您的指示，而无需您明确发出 执行此操作的命令。

顺便说一句，NodeJS (V8) 引擎试图通过使用单线程事件循环将事件分发到所有其他 I/O 和实用程序库线程来最小化争用和锁定，因此它们不必相互竞争访问事件和共享内存。 如您所见，NodeJS 取得了巨大的成功， 如果我们谈论的是特殊目的算法 你可以更进一步。 NodeJS architecture

阅读愉快。

Answer 4

如果一个线程等待而另一个线程执行它的操作，则几乎没有理由同步多个线程。人们可以使用 Executors 以更少的工作获得相同的结果，并且仍然感觉自己在玩线程。

public class Main {

    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        for(int i = 0; i<5;i++) {
            executor.submit(new PrintTask("performing job."));
            executor.submit(new PrintTask("perform some action"));
        }
        executor.shutdown();
    }

    private static class PrintTask implements Runnable {
        private String string;

        public PrintTask(String string) {
            this.string = string;
        }

        @Override
        public void run() {
            System.out.println(string);
        }
    }

}