thread_Semaphore信号量

Semaphore也是一个线程同步的辅助类，可以维护当前访问自身的线程个数，并提供了同步机制。
使用Semaphore可以控制同时访问资源的线程个数，例如，实现一个文件允许的并发访问数。

一个计数信号量。从概念上讲，信号量维护了一个许可集。
如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。
但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。拿到信号量的线程可以进入代码，否则就等待。

主要方法摘要：
void acquire():从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断
void release():释放一个许可，将其返回给信号量
int availablePermits():返回此信号量中当前可用的许可数
boolean hasQueuedThreads():查询是否有线程正在等待获取
1.简单例子

 维护当前访问自身的线程个数
    @Test

    public void semaphore1Test() throws InterruptedException {

        // 线程池

        ExecutorService exec = Executors.newCachedThreadPool();

        // 只能5个线程同时访问

        final Semaphore semp = new Semaphore(5);

        // 模拟20个客户端访问

        for (int index = 0; index < 20; index++) {

            final int NO = index;

            Runnable run = new Runnable() {

                public void run() {

                    try {

                        // 请求获得许可，如果有可获得的许可则继续往下执行，许可数减1。否则进入阻塞状态

                        semp.acquire();

                        System.out.println("Accessing: " + NO);

                        Thread.sleep((long) (Math.random() * 10000));

                        // 访问完后，释放许可，许可数加1，如果屏蔽下面的语句，则在控制台只能打印5条记录，之后线程一直阻塞

                        semp.release();

                         System.out.println("线程" + Thread.currentThread().getName() +  "已离开，当前已有" + (3-semp.availablePermits()) + "个并发");

                    } catch (InterruptedException e) {

                    }

                }

            };

            exec.execute(run);

        }

        // 退出线程池

        exec.shutdown();

    }

2.当锁

当信号量的数量上限是1时，Semaphore可以被当做锁来使用。通过acquire和release方法来保护关键区域。

    @Test

    public void semaphore2Test() throws InterruptedException {

        final Business business = new Business();

        ExecutorService executor = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 3; i++) {

            executor.execute(new Runnable() {

                public void run() {

                    business.service();

                }

            }

            );

        }

        executor.shutdown();

    }

   class Business {

        private int count;

        Lock lock = new ReentrantLock();

        Semaphore sp = new Semaphore(1);

        public void service() {

            // lock.lock();

            try {

                sp.acquire(); // 当前线程使用count变量的时候将其锁住，不允许其他线程访问

            } catch (InterruptedException e1) {

                e1.printStackTrace();

            }

            try {

                count++;

                try {

                    Thread.sleep(1000);

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

                System.out.println(count);

            } catch (RuntimeException e) {

                e.printStackTrace();

            } finally {

                // lock.unlock();

                sp.release(); // 释放锁

            }

        }

    }

3.生产者消费者模型，生成阻塞队列

a.消息通道

class SemaphoreQueue {

    private List<Integer> valueList;

    private Semaphore putActionNum;// 可以进行put操作的许可数量

    private Semaphore getActionNum;// 可以进行take操作的许可数量

    private Semaphore mutex;       //相当于锁 控制非线程安全的valueList的操作

    public SemaphoreQueue(int capacity) {

        putActionNum = new Semaphore(capacity);// 维护队列大小

        getActionNum = new Semaphore(0);       // 初始化时，队列为空，put操作许可数量为0

        mutex = new Semaphore(1);              // 用于保护非线程安全的valueList操作，用于并发生产时控制

        valueList = new ArrayList<Integer>(capacity);

    }

    public void put(Integer message) {

        try {

            putActionNum.acquire();// put操作许可减1

            mutex.acquire();

            valueList.add(message);

            mutex.release();

            getActionNum.release();// get操作许可加1

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

    }

    public Integer take() {

        Integer message = null;

        try {

            getActionNum.acquire();// get操作许可减1

            mutex.acquire();

            if (valueList.size() > 0) {

                message = valueList.get(0);

                valueList.remove(0);

            } else {

                return null;

            }

            mutex.release();

            putActionNum.release();// put操作许可加1

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        return message;

    }

}

b. 生产者和消费者

class Productor extends Thread {

    SemaphoreQueue queue;

    public Productor(SemaphoreQueue queue) {

        this.queue = queue;

    }

    public void run() {

        int i = 0;

        try {

            while (true) {

                i++;

                Integer message = new Integer(i);

                queue.put(message);

                if (i % 20 == 0) {

                    System.out.println("======== " + this.getName() + " 累计生产了 " + i + " 条消息  =======");

                    Thread.currentThread().sleep(1000);

                }

            }

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

class Cousumertor extends Thread {

    SemaphoreQueue queue;

    public Cousumertor(SemaphoreQueue queue) {

        this.queue = queue;

    }

    public void run() {

        try {

            while (true) {

                Integer message = queue.take();

                if (message != null) {

                    System.out.println("======== " + this.getName() + " 消费消息：" + message + " =======");

                }

                Thread.currentThread().sleep(100);

            }

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

c.测试

    public static void main(String[] args) {

        SemaphoreQueue queue = new SemaphoreQueue(20);

        // 开始生产

        Productor productor = new Productor(queue);

        productor.setName("生产者");

        productor.start();

        // 开始消费

        Cousumertor c1 = new Cousumertor(queue);

        c1.setName("消费者-c1");

        Cousumertor c2 = new Cousumertor(queue);

        c2.setName("消费者-c2");

        c1.start();

        c2.start();

    }

4. Semaphore vs. CountDownLatch
相同点：
两者都是用于线程同步的工具类，都通过定义了一个继承AbstractQueuedSynchronizer的内部类Sync来实现具体的功能.
不同点：
　　a. Semaphore提供了公平和非公平两种策略, 而CountDownLatch则不具备.
　　b. CountDownLatch：一个或者是一部分线程，等待另外一部线程都完成操作。
　　 Semaphorr: 维护一个许可集.通常用于限制可以访问某些资源（物理或逻辑的）的线程数目.

　　c. CountDownLatch中计数是不能被重置的。CountDownLatch适用于一次同步。当使用CountDownLatch时，任何线程允许多次调用countDown(). 那些调用了await()方法的线程将被阻塞，直到那些没有被阻塞线程调用countDown()使计数到达0为止。
　　Semaphore允许线程获取许可, 未获得许可的线程需要等待.这样防止了在同一时间有太多的线程执行.Semaphore的值被获取到后是可以释放的，并不像CountDownLatch那样一直减到0。

　　d. 使用CountDownLatch时，它关注的一个线程或者多个线程需要在其它在一组线程完成操作之后，在去做一些事情。比如：服务的启动等。使用Semaphore时,它关注的是某一个资源最多同时能被几个线程访问.

巴特西

thread_Semaphore信号量

最新文章

热门文章