Java - 被线程难题难住了

Question

基本上，我正在尝试实现一种机制，让两个线程并行运行。 Thread1 不断更新计数器值。当计数器值达到特定值的增量（例如 100、250、500 的倍数）时，我希望 Thread2 并行执行在计数器值上选择的特定任务。 Thread1 应继续计数，但如果 Thread2 尚未完成其任务，则不应超过键值。

用例：线程 1 已将计数器更新为 100。这将调度 Thread2 以执行 TaskA。 Thread1 仍在计数。计数器达到 250。如果 Thread2 已完成其任务，则 Thread1 应继续。否则，Thread1 应等待 TaskA 完成后再继续。

|t2             |t1   
|               | 
|               | 
|               | 
______100________ <----start thread 2 real quick
|               | 
|               | 
|               | 
|               | 
|               | 
|               | 
_______250______ <------at this point just wait for taskA to finish
|               |          IF it's not finished. If it is, start taskB and
|               |          continue counting
V               V

我一直在研究这个问题，但到目前为止我已经放弃了所有内容。我很感激代码/伪代码/提示/建议。提前致谢

Answer 1

CyclicBarrier 可用于创建线程等待另一个线程的屏障。因此，下面有两个线程“countingThread”和“taskThread”。 'countingThread' 将执行其计数，并在计数达到特定点时调用'await'（下面的方法-'checkBarrierCondition'）。

根据问题中的示例，当计数线程达到 100 时，它可以在屏障上调用“等待”，如果此时任务线程已完成其任务，则屏障将捕捉并且两者都会继续到下一个活动。如果任务还没有完成，那么计数器线程将等待执行任务的线程。

所有的锁定都由 CyclicBarrier 和并发框架处理

public class Threading {

public void execute() {
    final CyclicBarrier barrier = new CyclicBarrier(2);

    Thread countingThread = new Thread(new Tasker(barrier));
    Thread taskThread = new Thread(new Counter(barrier));

    countingThread.start();
    taskThread.start();

    try {
        countingThread.join();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

public static void main(String[] args) {

    new Threading().execute();

}

class Tasker implements Runnable {
    private CyclicBarrier barrier;

    Tasker(CyclicBarrier barrier) {
        this.barrier = barrier;
    }

    public void run() {
        String task = "taskA";      //just some mock-up task name

        while (!allTasksDone(task)) {
            task = performTask(task);
            try {
                System.out.println("Tasker : Await on barrier ");
                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }

        }
    }

}

class Counter implements Runnable {
    private CyclicBarrier barrier;

    Counter(CyclicBarrier barrier) {
        this.barrier = barrier;
    }

    public void run() {
        int counter = 0;  //just for the sake of example; starting at 0

        while (!isCountingDone(counter)) {
            counter = performCounting(counter);
            if (checkBarrierCondition(counter)) {
                try {
                    System.out.println("Counter : Await on barrier ");
                    barrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

}

Answer 2

您可能想使用锁？考虑一下 - 计数器：

import java.util.concurrent.locks.Lock;


public class ThreadOne extends Thread {

    private ThreadTwo two;
    private Lock lock;

    public ThreadOne(Lock l, ThreadTwo two) {
        this.two = two;
        this.lock = l;
        this.start();

    }

    @Override
    public void run() {
        int i = 0;
        while(true) {
            if(i%100==0) {
                // tell other thread to start
                two.startRunning();
                while(two.pending()) {
                    // wait until it actually started
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            // acquire the lock (or wait)
            lock.lock();
            try {
                // count up
                i++;
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        } 
    }

}

执行线程：

import java.util.concurrent.locks.Lock;


public class ThreadTwo extends Thread {

    private boolean pending = false;
    private Lock lock;

    public ThreadTwo(Lock l) {
        this.lock = l;
        this.start();
    }

    public void startRunning() {
        pending = true;
    }

    public boolean pending() {
        return pending;
    }

    @Override
    public void run() {
        while(true) {
            try {
                Thread.sleep(200);
            } catch (Exception e) {
            }
            if(pending) {
                lock.lock(); 
                try {
                    pending = false;
                    execute();
                } catch (Exception e) {
                } finally {
                    lock.unlock();
                }
            }
        }
    }

    private void execute() {
    }

}

以及如何启动它们。

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


public class Main {

    public static void main(String[] args) {
        Lock l = new ReentrantLock();
        ThreadTwo two = new ThreadTwo(l);
        ThreadOne one = new ThreadOne(l,two);
    }

}

Answer 3

package testRandomStuff;


public class ThreadingPuzzle {
    public int countMax = 25;
    public int factor = 5;
    public Thread threadA, threadB;

    private class Signal {
        public volatile boolean flag = true;

        public Signal(boolean initial) {
            flag = initial;
        }

        public synchronized void setFlag() {
            flag = true;
            notifyAll();
        }

        public synchronized void unsetFlag() {
            flag = false;
            notifyAll();
        }

        public synchronized boolean getFlag() {
            return flag;
        }
    }

    public Signal checkpoint = new Signal(true);
    public Signal doWork = new Signal(false);

    Runnable threadARunnable = new Runnable() {
        @Override
        public void run() {
            for (int i = 0; i < countMax; i++) {
                if (i % factor == 0) {
                    if (checkpoint != null) {
                        // --------mechanism to wait for threadB to finish---------
                        synchronized (checkpoint) {
                            try {
                                // -----use while loop to prevent spurious wakeup------
                                // Checkpoint flag is true in the first iteration, no need to wait.
                                while (!checkpoint.getFlag()) {
                                    checkpoint.wait();
                                }
                            } catch (InterruptedException ie) {
                                // handle exception
                            }
                        }
                        // ThreadB has finished last job when threadA leaves the above sync-block
                    }

                    // ------ start threadB real quick---------
                    // unset checkpoint flag, so that threadA will not proceed the next
                    // interation without threadB setting the flag first.
                    // send signal to threadB to wake it up
                    checkpoint.unsetFlag();
                    doWork.setFlag();

                }
                System.out.println("Thread A - count:"+i);
            }

        }
    };

    Runnable threadBRunnable = new Runnable() {
        @Override
        public void run() {
            while (true) {
                // --------mechanism to wait for threadA send job---------
                synchronized (doWork) {
                    try {
                        // -----use while loop to prevent spurious wakeup------
                        // doWork flag is false in the first iteration, wait for ThreadA.
                        while (!doWork.getFlag()) {
                            doWork.wait();
                        }
                    } catch (InterruptedException ie) {
                        // handle exception
                    }
                }

                doWork.unsetFlag();
                // -----------do what ever you need to do in threadB-----------
                System.out.println("Thread B - do some work");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {

                }
                System.out.println("Thread B - done working");
                // ------------Finish work, notify threadA---------
                checkpoint.setFlag();
            }
        }

    };

    public ThreadingPuzzle() {
        // FIXME Auto-generated constructor stub
    }

    public static void main(String[] args){
        ThreadingPuzzle puzzle = new ThreadingPuzzle();
        puzzle.threadA = new Thread(puzzle.threadARunnable);
        puzzle.threadB = new Thread(puzzle.threadBRunnable);
        puzzle.threadA.start();
        puzzle.threadB.start();
    }

}


SIMULATION RESULTS
Thread B - do some work
Thread A - count:0
Thread A - count:1
Thread A - count:2
Thread A - count:3
Thread A - count:4
Thread B - done working
Thread B - do some work
Thread A - count:5
Thread A - count:6
Thread A - count:7
Thread A - count:8
Thread A - count:9
Thread B - done working
Thread B - do some work
Thread A - count:10
Thread A - count:11
Thread A - count:12
Thread A - count:13
Thread A - count:14
Thread B - done working
Thread B - do some work
Thread A - count:15
Thread A - count:16
Thread A - count:17
Thread A - count:18
Thread A - count:19
Thread B - done working
Thread B - do some work
Thread A - count:20
Thread A - count:21
Thread A - count:22
Thread A - count:23
Thread A - count:24
Thread B - done working
Thread B - do some work
Thread B - done working

Answer 4

我建议看看 Java 的执行器服务。它确实抽象了与多线程相关的大部分复杂性。另外，如果将来需要，您可以轻松增加执行任务的线程数。基本上你在你的第一个线程中运行计数。当您想在另一个线程中执行任务时，您只需创建一个可调用对象。 API 将为您的可调用对象返回一个未来。当您在线程 1 中完成处理/计数后，您只需从线程 1 调用 get 或 getValue 即可。现在这样做的好处是，如果其他线程完成处理，它将立即返回结果。如果其他线程正忙于处理该任务，那么它将阻塞您的 thread1，直到返回结果。请注意，您不需要手动进行任何锁定、阻止或通知。如果您在多个线程之间共享数据，请不要忘记使用线程安全集合。希望这会有所帮助！