同步三个线程

Question

在一次采访中被问到这个问题，试图解决它......但没有成功。 我想到了使用 CyclicBarrier

共有三个线程 T1 打印 1,4,7... T2 打印 2,5,8... 而 T3 打印 3,6,9 ...。怎么同步这三个来打印序列1,2,3,4,5,6,7,8,9....

我尝试编写和运行以下代码

public class CyclicBarrierTest {
    public static void main(String[] args) {
        CyclicBarrier cBarrier = new CyclicBarrier(3);
        new Thread(new ThreadOne(cBarrier,1,10,"One")).start();
        new Thread(new ThreadOne(cBarrier,2,10,"Two")).start();
        new Thread(new ThreadOne(cBarrier,3,10,"Three")).start();
    }
}

class ThreadOne implements Runnable {
    private CyclicBarrier cb;
    private String name;
    private int startCounter;
    private int numOfPrints;

    public ThreadOne(CyclicBarrier cb, int startCounter,int numOfPrints,String name) {
        this.cb = cb;
        this.startCounter=startCounter;
        this.numOfPrints=numOfPrints;
        this.name=name;
    }

    @Override
    public void run() {
        for(int counter=0;counter<numOfPrints;counter++)
        {
            try {
            // System.out.println(">>"+name+"<< "+cb.await());
            cb.await();
            System.out.println("["+name+"] "+startCounter);
            cb.await();
            //System.out.println("<<"+name+">> "+cb.await());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        startCounter+=3;
        }
    }

}

输出

[Three] 3
[One] 1
[Two] 2
[One] 4
[Two] 5
[Three] 6
[Two] 8
[One] 7
[Three] 9
[One] 10
[Two] 11
[Three] 12
[Two] 14
[One] 13
[Three] 15
[One] 16
[Two] 17
[Three] 18
[Two] 20
[One] 19
[Three] 21
[One] 22
[Two] 23
[Three] 24
[Two] 26
[One] 25
[Three] 27
[One] 28
[Two] 29
[Three] 30

谁能帮我正确回答？

类似的问题 Thread Synchronization - Synchronizing three threads to print 012012012012..... not working

Answer 1

我已经使用事件编写了一个三线程应用程序

#include <iostream>
#include <Windows.h>
#include <atomic>
using namespace std;


HANDLE firstThreadEvent = CreateEvent(NULL, true, true, NULL);
HANDLE secondThreadEvent = CreateEvent(NULL, true, false, NULL);
HANDLE thirdThreadEvent = CreateEvent(NULL, true, false, NULL);
std::atomic<int> m_int = 1;

DWORD WINAPI firstThreadFun(LPVOID lparam)
{
    while (1)
    {
        ::WaitForSingleObject(firstThreadEvent,INFINITE);
        cout << "By first thread " << m_int << std::endl;
        m_int++;
        ResetEvent(firstThreadEvent);
        SetEvent(secondThreadEvent);

    }
}
DWORD WINAPI secondThreadFun(LPVOID lparam)
{
    while (1)
    {
        ::WaitForSingleObject(secondThreadEvent, INFINITE);
        cout << "By second thread "<< m_int << std::endl;
        m_int++;
        ResetEvent(secondThreadEvent);
        SetEvent(thirdThreadEvent);

    }

}
DWORD WINAPI thirdThreadFun(LPVOID lparam)
{
    while (1)
    {
        ::WaitForSingleObject(thirdThreadEvent, INFINITE);
        cout << "By third thread " << m_int << std::endl;
        m_int++;
        ResetEvent(thirdThreadEvent);
        SetEvent(firstThreadEvent);

    }
}



int main()
{
    HANDLE hnd[3];
    hnd[0] = CreateThread(NULL, 0, &firstThreadFun, NULL, 0, NULL);
    hnd[1] = CreateThread(NULL, 0, &secondThreadFun, NULL, 0, NULL);
    hnd[2] = CreateThread(NULL, 0, &thirdThreadFun, NULL, 0, NULL);

    WaitForMultipleObjects(3, hnd, true, INFINITE);
    CloseHandle(hnd[0]);
    CloseHandle(hnd[1]);
    CloseHandle(hnd[2]);
    return 0;
}

Answer 2

因为线程应该被赋予相同的权重，即第一个之后第二个第三个。以下工作。

public class MultiThreadSynchronize {

static int index = 0;
static Object lock = new Object();

static class NumberPrinter extends Thread {
    private final int remainder;
    private final int noOfThreads;
    private int value;

    public NumberPrinter(String name, int remainder, int noOfThreads, int value) {
        super(name);
        this.remainder = remainder;
        this.noOfThreads = noOfThreads;
        this.value = value;
    }

    @Override
    public void run() {
        while (index < 20) {
            synchronized (lock) {
                while ((index % noOfThreads) != remainder) {. // base condition where all threads except one waits.
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                index++;
                System.out.println(getName() + " " + value);
                value += 3;
                lock.notifyAll();
            }
        }
    }
}

public static void main(String[] args) {
    NumberPrinter numberPrinter1 = new NumberPrinter("First Thread", 0, 3, 1);
    NumberPrinter numberPrinter2 = new NumberPrinter("Second Thread", 1, 3, 2);
    NumberPrinter numberPrinter3 = new NumberPrinter("Third Thread", 2, 3, 3);
    numberPrinter1.start(); numberPrinter2.start(); numberPrinter3.start();
}
}

Answer 3

这是一种方法，它适用于使用同步、等待和 notifyAll 的任意数量的线程。
turn 变量控制必须执行的线程。这样的线程执行任务，增加turn（以线程数为模），通知所有线程并进入while循环等待，直到再次轮到它。

public class Test2 {

    final static int LOOPS = 10;
    final static int NUM_TREADS = 3;

    static class Sequenced extends Thread {

        static int turn = 0;
        static int count = 0;
        static Object lock = new Object();
        final int order;

        public Sequenced(int order) {
            this.order = order;
        }

        @Override
        public void run() {
            synchronized (lock) {
                try {
                    for (int n = 0; n < LOOPS; ++n) {
                        while (turn != order) {
                            lock.wait();
                        }
                        ++count;
                        System.out.println("T" + (order + 1) + " " + count);
                        turn = (turn + 1) % NUM_TREADS;
                        lock.notifyAll();
                    }
                } catch (InterruptedException ex) {
                    // Nothing to do but to let the thread die.
                }
            }
        }
    }

    public static void main(String args[]) throws InterruptedException {
        Sequenced[] threads = new Sequenced[NUM_TREADS];
        for (int n = 0; n < NUM_TREADS; ++n) {
            threads[n] = new Sequenced(n);
            threads[n].start();
        }
        for (int n = 0; n < NUM_TREADS; ++n) {
            threads[n].join();
        }
    }

}

Answer 4

正如其他人已经提到的，CyclicBarrier 并不是完成这项任务的最佳工具。

我也同意解决方案是链接线程并让一个线程为下一个线程设置执行的观点。

这是一个使用信号量的实现：

import java.util.concurrent.BrokenBarrierException; 
import java.util.concurrent.Semaphore;

public class PrintNumbersWithSemaphore implements Runnable {

private final Semaphore previous;

private final Semaphore next;

private final int[] numbers;

public PrintNumbersWithSemaphore(Semaphore previous, Semaphore next, int[] numbers) {
    this.previous = previous;
    this.next = next;
    this.numbers = numbers;
}

@Override
public void run() {

    for (int i = 0; i < numbers.length; i++) {
        wait4Green();

        System.out.println(numbers[i]);

        switchGreen4Next();
    }
}

private void switchGreen4Next() {
        next.release();
}

private void wait4Green() {
    try {
        previous.acquire();
    } catch (InterruptedException e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }
}

static public void main(String argv[]) throws InterruptedException, BrokenBarrierException {
    Semaphore sem1 = new Semaphore(1);
    Semaphore sem2 = new Semaphore(1);
    Semaphore sem3 = new Semaphore(1);
    sem1.acquire();
    sem2.acquire();
    sem3.acquire();
    Thread t1 = new Thread(new PrintNumbersWithSemaphore(sem3, sem1, new int[] { 1, 4, 7 }));
    Thread t2 = new Thread(new PrintNumbersWithSemaphore(sem1, sem2, new int[] { 2, 5, 8 }));
    Thread t3 = new Thread(new PrintNumbersWithSemaphore(sem2, sem3, new int[] { 3, 6, 9 }));
    t1.start();
    t2.start();
    t3.start();
    sem3.release();

    t1.join();
    t2.join();
    t3.join();
}

}

这是另一个，在我看来，使用 CyclicBarrier 实现相当麻烦：

import java.util.concurrent.BrokenBarrierException; 
import java.util.concurrent.CyclicBarrier;

public class PrintNumbersWithCyclicBarrier implements Runnable {

private final CyclicBarrier previous;

private final CyclicBarrier next;

private final int[] numbers;

public PrintNumbersWithCyclicBarrier(CyclicBarrier previous, CyclicBarrier next, int[] numbers) {
    this.previous = previous;
    this.next = next;
    this.numbers = numbers;
}

@Override
public void run() {

    for (int i = 0; i < numbers.length; i++) {
        wait4Green();

        System.out.println(numbers[i]);

        switchRed4Myself();

        switchGreen4Next();
    }
}

private void switchGreen4Next() {
    try {
        next.await();
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }
}

private void switchRed4Myself() {
    previous.reset();
}

private void wait4Green() {
    try {
        previous.await();
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }
}

static public void main(String argv[]) throws InterruptedException, BrokenBarrierException {
    CyclicBarrier cb1 = new CyclicBarrier(2);
    CyclicBarrier cb2 = new CyclicBarrier(2);
    CyclicBarrier cb3 = new CyclicBarrier(2);
    Thread t1 = new Thread(new PrintNumbersWithCyclicBarrier(cb3, cb1, new int[] { 1, 4, 7 }));
    Thread t2 = new Thread(new PrintNumbersWithCyclicBarrier(cb1, cb2, new int[] { 2, 5, 8 }));
    Thread t3 = new Thread(new PrintNumbersWithCyclicBarrier(cb2, cb3, new int[] { 3, 6, 9 }));
    t1.start();
    t2.start();
    t3.start();
    cb3.await();

    t1.join();
    t2.join();
    t3.join();
}

}

Answer 5

是否要求使用单个 CyclicBarrier？我的建议是：

1. 为每个 ThreadOne 实例分配两个 CyclicBarriers

2. 你应该创建一个循环图，这样

   ThreadOne_1 -> ThreadOne_2 -> ThreadOne_3 -> ThreadOne_1 -> etc...

3. 要实现 (2)，您需要将父级的 CyclicBarrier 与子级共享，最后一个任务应与第一个线程的子级共享其 CB。

回答您的问题：

试图浏览文档，但不是很清楚 await() 到底做了什么......

Await 将自行挂起，直到 N 线程数在屏障上调用 await。因此，如果您定义 new CyclicBarrier(3) 超过一次 3 个线程调用 await，则屏障将允许线程继续。

什么时候使用reset()

你不需要，一旦线程数到达，它会自动跳闸