Basic Of Concurrency(十七: Semaphores)

Semaphores是一个用于保证线程间相互发送信号且信号不会丢失的同步结构,与Lock相似,一样可以保证临界区的安全访问.在Java5java.util.concurrent包中已有相关实现,所以咱们不须要本身手动实现.但仍然颇有有必要知道怎么使用它们以及底层原理.

简单的Semaphore

如下是一个简单的Semaphores实现:

public class Semaphore {
    private boolean signal = false;

    public synchronized void take() {
        signal = true;
        notify();
    }

    public synchronized void release() {
        while (!signal) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal = false;
    }
}
复制代码

在Semaphore中,调用take()发送信号会将信号先存储起来.而调用release()则是等待信号.当收到信号时会从新清除信号并退出release()方法.

像这样调用Semaphore可以帮助咱们解决信号丢失问题.你可使用take()和release()来代替wait()和notify()方法.若是在调用release()以前就已经调用了take(),再去调用release()仍然可以知道take()被调用过,由于take()调用会将信号存储在signal变量中.而不像wait()和notify()那样.

take()和release()方法的命名是有渊源的.它们的名字来源于将semaphores当成锁来用的时候.稍后会做解释.

使用Semaphore通信

这是一个简单的实例,两个线程使用Semaphore互相通信.

public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore();
        Sender sender = new Sender(semaphore);
        Receiver receiver = new Receiver(semaphore);
    }
}
复制代码

public class Sender extends Thread{
    private Semaphore semaphore;

    public Sender(Semaphore semaphore){
        this.semaphore = semaphore;
    }

    @Override
    public void run(){
        while (true){
            // do something then send signal
            semaphore.take();
        }
    }
}
复制代码

public class Receiver extends Thread {
    private Semaphore semaphore;

    public Receiver(Semaphore semaphore) {
        this.semaphore = semaphore;
    }

    @Override
    public void run() {
        while (true) {
            semaphore.release();
            // receive signal then do something
        }
    }
}
复制代码

CountingSemaphore

在前文给出的Semaphore实现并无记录take()调用了多少次即发送了多少个信号.咱们能够作出小小的改动来实现这个功能.咱们称这种实现为CountingSemaphore:

public class CountingSemaphore {
    private int signal = 0;

    public synchronized void take() {
        signal++;
        notify();
    }

    public synchronized void release() {
        while (signal == 0) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal--;
    }
}
复制代码

BoundedSemaphore

CountingSemaphore并无设置它所能存储信号数量的上限.咱们能够作出小小改动来实现这个功能.以下:

public class BoundedSemaphore {
    private int signal = 0;
    private int bound;

    public BoundedSemaphore(int upperBound) {
        this.bound = upperBound;
    }

    public synchronized void take() {
        while (signal == bound) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal++;
        notify();
    }

    public synchronized void release() {
        while (signal == 0) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal--;
        notify();
    }
}
复制代码

咱们能够注意到在信号达到上限时,调用take()会发生阻塞进入等待状态.BoundedSemaphore在到达信号上限时会阻塞take()调用直到有线程调用release()减小信号数量为止.

将BoundedSemaphore当成锁来使用

咱们可以将BoundedSemaphore当成锁来使用.咱们只须要将信号上限设置为1,将临界区代码经过take()和release()调用包裹起来便可.下面是一个示例:

BoundedSemaphore semaphore = new BoundedSemaphore(1);
semaphore.take();
try{
  //critical section
} finally {
  semaphore.release();
}
复制代码

上面示例中,take()和release()的调用在同一个线程中完成.当一个线程取得Semaphore实例时,其余线程将会被阻塞在take()调用上,直到持有Semaphore实例的线程调用release()为止.当线程调用take()退出后能够正常调用release()而不会进入阻塞.

此外你还可使用BoundedSemaphore的上限来限制同时进入临界区的线程数量.在上面实例中,咱们能够限制BoundedSemaphore的上限为5.那么5个线程能够同时访问临界区代码,前提是这5个线程的操做不会有线程安全问题,否则你的应用将会产生不可预期的后果.

咱们将release()调用放置在finally语句里面,用于避免临界区代码抛出异常而不能释放信号.

该系列博文为笔者复习基础所著译文或理解后的产物,复习原文来自Jakob Jenkov所著Java Concurrency and Multithreading Tutorial

上一篇: Reentrance Lockout
下一篇: 阻塞队列