java - 从另一个线程暂停线程，并从另一个线程停止/启动它

Question

我已经找到good hints here 。但我有更困难的任务 - 额外的要求是:
-我的低优先级永久线程可以从主线程启动/停止(这是相同的)-但它还必须锁定一个资源以进行独占访问。
-我的低优先级永久线程可以从另一个高优先级线程暂停/继续(然后它们也锁定并使用该资源)
-我还希望低优先级线程不要在每个循环中释放锁，而只在被告知时释放锁(出于速度目的 - 因为我需要初始化/取消初始化资源)(如果我有在每个循环中释放锁，然后我可以让java来管理竞争线程并期望高优先级线程时不时地获胜)。

我想出了解决方案，我认为它是安全可靠的。但我是新手，所以我愿意在我的解决方案中查找错误或改进建议。所以我的问题是:
1)我的解决方案在所有情况下都真的是线程安全的吗？
2)这是最佳解决方案还是我可以改进一些东西？

如果您同意，则将其用作模板。

核心代码如下:

FlagNotify lowPriorRunReq   = new FlagNotify(false); // low priority task run request
FlagNotify lowPriorPauseReq = new FlagNotify(false); // low priority task pause request (it uses notify)
volatile boolean lowPriorRunFlag = false;  // low priority task run flag
Lock      groupLock        = new ReentrantLock();  // group lock (used to acquire lowPriorRunFlag always correctly)
Semaphore resourceSemaphore = new Semaphore(1);    // main semaphore protecting resource that has to be accessed sequentially

public class PrioritySingleTaskThread extends Thread {
    @Override
    public void run() {
        prn("High Priority Task created");
        groupLock.lock();
        if(lowPriorRunFlag == true) lowPriorPauseReq.setNotify(true);
        resourceSemaphore.acquireUninterruptibly();
        groupLock.unlock();
        accessResource("high priority task");
        resourceSemaphore.release();
        groupLock.lock();
        if(lowPriorPauseReq.get() == true) lowPriorPauseReq.setNotify(false);
        groupLock.unlock();
        prn("High Priority Task closed");
    }
}

public class LowPriorityContinuousThread extends Thread {
    void getResourceSemaphore(){
        groupLock.lock();
        resourceSemaphore.acquireUninterruptibly();
        lowPriorRunFlag = true;
        groupLock.unlock();
        accessResource("low priority init"); // here it is initialization and I want to happen only on request from priority thread
    }
    void releaseResourceSemaphore(){
        accessResource("low priority de-init"); // here it is de-initialization and I want to happen only on request from priority thread
        lowPriorRunFlag = false;
        resourceSemaphore.release();
    }

    @Override
    public void run() {
        while(true){
            //prn("Low Priority Run req: "+lowPriorRunReq.get());
            if(lowPriorRunReq.get() == true && lowPriorRunFlag == false){
                prn("Low  Priority Task starting");
                getResourceSemaphore();
                prn("Low  Priority Task started");
            }
            if(lowPriorRunReq.get() == false && lowPriorRunFlag == true){
                prn("Low  Priority Task stopping");
                releaseResourceSemaphore();
                prn("Low  Priority Task stopped");
                lowPriorRunReq.smartWait(true);
            }
            // note keep checking lowPriorRunFlag. Imagine there is RunFlag detected by high priority thread
            // before de-asserted, then pauseReq would be requested from high priority thread
            // then resource is released when low priority task stops.
            // High priority lock and use resource, but since pauseReq is set
            // this thread would try to access device in order to de-init and pause (unless we check runFlag)
            if(lowPriorPauseReq.get() == true && lowPriorRunFlag == true){
                prn("Low  Priority Task pausing");
                releaseResourceSemaphore();
                prn("Low  Priority Task paused");
                lowPriorPauseReq.smartWait(false);
                getResourceSemaphore();
                prn("Low  Priority Task continue");
            }
            if(lowPriorRunFlag){
                accessResource("low priority task");
            }
        }
    }
}

这里是完整的可编译 Java 代码，包括测试平台(所以我暗示这是一个安全的解决方案 - 但你永远不知道这些线程)

import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Main {
    // this inner class is only for setting flag and waiting to it by notify - which does not hog CPU
    public class FlagNotify{
        private Boolean flag; // do not synchro on Boolean - it is immutable, thus new object is created every value change....
        private Object synchro = new Object();

        public FlagNotify(boolean val) {
            flag = val;
        }

        public void setNotify(boolean val) {
            synchronized (synchro) {
                flag = val;
                synchro.notify();
            }
        }

        public boolean get(){
            return flag;
        }
        public void smartWait(boolean expVal){
            synchronized (synchro){
                while(flag != expVal){
                    try {
                        synchro.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    FlagNotify lowPriorRunReq   = new FlagNotify(false); // low priority task run request
    FlagNotify lowPriorPauseReq = new FlagNotify(false); // low priority task pause request (it uses notify)
    volatile boolean lowPriorRunFlag = false;  // low priority task run flag
    Lock      groupLock        = new ReentrantLock();  // group lock (used to acquire lowPriorRunFlag always correctly)
    Semaphore resourceSemaphore = new Semaphore(1);    // main semaphore protecting resource that has to be accessed sequentially

    public class PrioritySingleTaskThread extends Thread {
        @Override
        public void run() {
            prn("High Priority Task created");
            groupLock.lock();
            if(lowPriorRunFlag == true) lowPriorPauseReq.setNotify(true);
            resourceSemaphore.acquireUninterruptibly();
            groupLock.unlock();
            accessResource("high priority task");
            resourceSemaphore.release();
            groupLock.lock();
            if(lowPriorPauseReq.get() == true) lowPriorPauseReq.setNotify(false);
            groupLock.unlock();
            prn("High Priority Task closed");
        }
    }

    public class LowPriorityContinuousThread extends Thread {
        void getResourceSemaphore(){
            groupLock.lock();
            resourceSemaphore.acquireUninterruptibly();
            lowPriorRunFlag = true;
            groupLock.unlock();
            accessResource("low priority init"); // here it is initialization and I want to happen only on request from priority thread
        }
        void releaseResourceSemaphore(){
            accessResource("low priority de-init"); // here it is de-initialization and I want to happen only on request from priority thread
            lowPriorRunFlag = false;
            resourceSemaphore.release();
        }

        @Override
        public void run() {
            while(true){
                //prn("Low Priority Run req: "+lowPriorRunReq.get());
                if(lowPriorRunReq.get() == true && lowPriorRunFlag == false){
                    prn("Low  Priority Task starting");
                    getResourceSemaphore();
                    prn("Low  Priority Task started");
                }
                if(lowPriorRunReq.get() == false && lowPriorRunFlag == true){
                    prn("Low  Priority Task stopping");
                    releaseResourceSemaphore();
                    prn("Low  Priority Task stopped");
                    lowPriorRunReq.smartWait(true);
                }
                // note keep checking lowPriorRunFlag. Imagine there is RunFlag detected by high priority thread
                // before de-asserted, then pauseReq would be requested from high priority thread
                // then resource is released when low priority task stops.
                // High priority lock and use resource, but since pauseReq is set
                // this thread would try to access device in order to de-init and pause (unless we check runFlag)
                if(lowPriorPauseReq.get() == true && lowPriorRunFlag == true){
                    prn("Low  Priority Task pausing");
                    releaseResourceSemaphore();
                    prn("Low  Priority Task paused");
                    lowPriorPauseReq.smartWait(false);
                    getResourceSemaphore();
                    prn("Low  Priority Task continue");
                }
                if(lowPriorRunFlag){
                    accessResource("low priority task");
                }
            }
        }
    }
//-------------------------------------------------------------------------
//-- following functions are meant only for testing
    AtomicInteger clashDetector = new AtomicInteger(0); // only for testing purposes

    public void accessResource(String from){
        prn("Resource used from "+from);
        if(clashDetector.addAndGet(1)>1){
            System.out.println("Clash detected - you are a bad programmer :((((((");
            System.exit(-1);
        }
        sleepRandom(5);
        clashDetector.getAndAdd(-1);
    }

    public void sleepRandom(long maxMiliSec){
        mySleep((long)(Math.random()*maxMiliSec));
    }

    public void mySleep(long miliSec){
        try{
            Thread.sleep(miliSec);
        }catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    void prn(String s){
        System.out.println(s);
    }

    public void test(){
        new LowPriorityContinuousThread().start();
        for(long i=0; i< (long)1e3; i++){
            lowPriorRunReq.setNotify(true);
            for(int j=0; j<Math.random()*100;j++){
                sleepRandom(10);
                new PrioritySingleTaskThread().start();
            }
            //sleepRandom(20);
            lowPriorRunReq.setNotify(false);

            for(int j=0; j<Math.random()*20;j++){
                sleepRandom(10);
                new PrioritySingleTaskThread().start();
            }
            //sleepRandom(20);
        }
        mySleep(200);
        System.out.println("Test OK :)))))))))))))))))))))");
        mySleep(200);
        System.exit(0);
    }

    public static void main(String[] args) throws Exception {
        new Main().test();
    }

}

Answer 1

我不知道“启动/停止”“永久线程”意味着什么，但可用于暂停线程的一种设计模式称为“十字转门”。当只有一个 Controller 线程被允许“锁定”或“解锁”旋转门时，以下是有效的:

import java.util.concurrent.Semaphore;

class Turnstile {
    private final Semaphore semaphore = Semaphore.new(1);

    // Called only from the "controller" thread.
    public void lock() {
        semaphore.acquire();
    }

    // Called only from the "controller" thread.
    public void unlock() {
        semaphore.release();
    }

    // Called from worker threads.
    public void passThrough() {
        semaphore.lock();
        semaphore.release();
    }
}

最初，旋转门处于“解锁”状态，当工作人员调用 passThrough() 方法时，它会立即返回。如果主线程“锁定”旋转栅门，则任何调用 passThrough() 的工作线程都将被阻塞，直到主线程再次“解锁”它。然后，所有的worker都会一一“通过”。

<小时/>

如果您想要拥有多个“主人”，您可以修改此示例，但是当一个主人希望锁定十字转门而另一个主人希望将其解锁时，由您决定如何解决冲突。

<小时/>

再说一次，而不是修改Turnstile ，你可以写一个新的 MultiMasterTurnstile处理冲突解决并使用 Turnstile真正阻止 worker 。