Guarded Suspension设计模式

Posted 2020-12-01 moris5013

Guarded Suspension 设计模式可以保证，当线程在访问某个对象时，发现条件不满足，就挂起等待条件满足时再次访问

public class GuardedSuspensionQueue {
    
    // 定义存放Integer类型的queue
    private final LinkedList<String> queue = new LinkedList<String>();

    // 定义queue的最大容量为5
    private  final int LIMIT = 5;

    public  static final Random random = new Random(System.currentTimeMillis());

    // 往queue中插入数据，如果queue中的元素超过了最大容量，则会陷入阻塞
    public void offer(String data) {
        synchronized (this) {
            // 判断queue的当前元素是否超过了LIMIT
            while (queue.size() >= LIMIT) {
                // 挂起当前线程，使其陷入阻塞
                System.out.println("向queue中添加元素，元素个数超过了限制，"+Thread.currentThread().getName()+"被挂起！");
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            // 插入元素并且唤醒take线程
            queue.addLast(data);
            this.notifyAll();
        }
    }

    
    // 从队列中获取元素，如果队列此时为空，则会使当前线程阻塞
    public String take()  {
        synchronized (this) {
            // 判断如果队列为空
            while (queue.isEmpty()) {
                // 则挂起当前线程
                System.out.println(Thread.currentThread().getName()+"从queue中取出元素，元素个数为0，被挂起！");
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                    return null;
                }
            }
            //removeFirst不会返回null值，queue没有元素会抛出异常
            String res =  queue.removeFirst();
            // 通知offer线程可以继续插入数据了
            this.notifyAll();
            return  res;
        }
    }

    
    public static void main(String[] args) throws InterruptedException {
        
        GuardedSuspensionQueue queue = new GuardedSuspensionQueue();
        
        TakerThread takerThread = new TakerThread(queue,"【线程A】");
        takerThread.start();
        
        Thread.sleep(2000);
        
        // 创建3个生产线程向queue中添加数据
        IntStream.rangeClosed(1, 3).forEach(i -> {
            new Thread(() -> {
                for (int j = 1; j <= 5; j++) {
                    queue.offer("【线程"+i+"】放入的"+j);
                }
                try {
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        });
        
        Thread.sleep(10_000);
        
        takerThread.close();
    }

}

class TakerThread extends Thread{
    private volatile boolean  close = false;
    private  GuardedSuspensionQueue   queue ;
    TakerThread(GuardedSuspensionQueue queue,String name){
        this.queue = queue;
        setName(name);
    }
    @Override
    public void run() {
        while(!close) {
            String item = queue.take();
            if(item == null) {
                break;
            }
            System.out.println(Thread.currentThread().getName() + "从queue中取出数据：" + item);
            try {
                Thread.sleep(GuardedSuspensionQueue.random.nextInt(1000));
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                break;
            }
        }
    }
    
    public void close() {
        this.close = close;
        this.interrupt();
    }
    
}