灯火互联
管理员
管理员
  • 注册日期2011-07-27
  • 发帖数41778
  • QQ
  • 火币41290枚
  • 粉丝1086
  • 关注100
  • 终身成就奖
  • 最爱沙发
  • 忠实会员
  • 灌水天才奖
  • 贴图大师奖
  • 原创先锋奖
  • 特殊贡献奖
  • 宣传大使奖
  • 优秀斑竹奖
  • 社区明星
阅读:3213回复:0

java5的Semaphere并发同步器

楼主#
更多 发布于:2012-09-08 09:41


ava.util.concurrent
类 Semaphore
public class Semaphoreextends Objectimplements Serializable一个计数信号量。
从概念上讲,信号量维护了一个许可集。
如有必要,在许可可用前会阻塞每一个 acquire(),然后再获取该许可。
每个 release() 添加一个许可,从而可能释放一个正在阻塞的获取者。
但是,不使用实际的许可对象,Semaphore 只对可用许可的号码进行计数,并采取相应的行动。
Semaphore 通常用于限制可以访问某些资源(物理或逻辑的)的线程数目。例如,下面的类使用信号量控制对内容池的访问:
class Pool {
     private static final int MAX_AVAILABLE = 100;
    private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);
    public Object getItem() throws interruptedException {
          available.acquire();
          return getNextAvailableItem();
  }
   public void putItem(Object x) {
     if (markAsUnused(x))
        available.release();
   }
   // Not a particularly efficient data structure; just for demo
   protected Object[] items =   whatever kinds of items being managed
   protected boolean[] used = new boolean[MAX_AVAILABLE];

   protected synchronized Object getNextAvailableItem() {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (!used) {
          used = true;
          return items;
       }
     }
     return null; // not reached
   }
   protected synchronized boolean markAsUnused(Object item) {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (item == items) {
          if (used) {
            used = false;
            return true;
          } else
            return false;
       }
     }
     return false;
   }
}
获得一项前,每个线程必须从信号量获取许可,从而保证可以使用该项。
该线程结束后,将项返回到池中并将许可返回到该信号量,从而允许其他线程获取该项。
注意,调用 acquire() 时无法保持同步锁,因为这会阻止将项返回到池中。
信号量封装所需的同步,以限制对池的访问,这同维持该池本身一致性所需的同步是分开的。
将信号量初始化为 1,使得它在使用时最多只有一个可用的许可,从而可用作一个相互排斥的锁。
这通常也称为二进制信号量,因为它只能有两种状态:一个可用的许可,或零个可用的许可。
按此方式使用时,二进制信号量具有某种属性(与很多 Lock 实现不同),即可以由线程释放“锁”,而不是由所有者(因为信号量没有所有权的概念)。
在某些专门的上下文(如死锁恢复)中这会很有用。
此类的构造方法可选地接受一个公平 参数。当设置为 false 时,此类不对线程获取许可的顺序做任何保证。
特别地,闯入 是允许的,也就是说可以在已经等待的线程前为调用 acquire() 的线程分配一个许可,
从逻辑上说,就是新线程将自己置于等待线程队列的头部。
当公平设置为 true 时,信号量保证对于任何调用获取方法的线程而言,都按照处理它们调用这些方法的顺序(即先进先出;FIFO)来选择线程、获得许可。
注意,FIFO 排序必然应用到这些方法内的指定内部执行点。
所以,可能某个线程先于另一个线程调用了 acquire,但是却在该线程之后到达排序点,并且从方法返回时也类似。
还要注意,非同步的 tryAcquire 方法不使用公平设置,而是使用任意可用的许可。
通常,应该将用于控制资源访问的信号量初始化为公平的,以确保所有线程都可访问资源。
为其他的种类的同步控制使用信号量时,非公平排序的吞吐量优势通常要比公平考虑更为重要。
此类还提供便捷的方法来同时 acquire 和释放多个许可。
小心,在未将公平设置为 true 时使用这些方法会增加不确定延期的风险。
内存一致性效果:线程中调用“释放”方法(比如 release())之前的操作 happen-before 另一线程中紧跟在成功的“获取”方法(比如 acquire())之后的操作。
acquire
public void acquire()
             throws InterruptedException从此信号量获取一个许可,在提供一个许可前一直将线程阻塞,否则线程被中断。
获取一个许可(如果提供了一个)并立即返回,将可用的许可数减 1。
如果没有可用的许可,则在发生以下两种情况之一前,禁止将当前线程用于线程安排目的并使其处于休眠状态:
某些其他线程调用此信号量的 release() 方法,并且当前线程是下一个要被分配许可的线程;或者
其他某些线程中断当前线程。
如果当前线程:
被此方法将其已中断状态设置为 on ;或者
在等待许可时被中断。
则抛出 InterruptedException,并且清除当前线程的已中断状态。
抛出:
InterruptedException - 如果当前线程被中断
release
public void release()释放一个许可,将其返回给信号量。
释放一个许可,将可用的许可数增加 1。
如果任意线程试图获取许可,则选中一个线程并将刚刚释放的许可给予它。然后针对线程安排目的启用(或再启用)该线程。
不要求释放许可的线程必须通过调用 acquire() 来获取许可。通过应用程序中的编程约定来建立信号量的正确用法。
应用实例:
package com.itm.thread;
/*
*

Semaphore的作用类似Lock的功能,不同的是Semaphore的构造函数中可以传入一个int型的参数,用来确定创建一个多大的通道。
Lock一次只允许一个线程进入,解锁后才允许别的线程进入。而Semaphore可以允许多个线程同时进入,一旦有线程释放就会空出一个位置让另外的线程进入。
相当与Lock是一个单车道,而Semaphore是一个多车道。

*/
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class SemaphoreTest {
    public static void main(String[] args) {
        ExecutorService es = Executors.newCachedThreadPool();
      
        // final Semaphore sema=new Semaphore(3); //共有3个通道
        final Semaphore sema = new Semaphore(3, true); // 表示按排队的先后顺序进入
        for (int i = 0; i < 10; i++) {
            es.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        sema.acquire();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("线程" + Thread.currentThread().getName()
                            + "已进入,当前有" + (3 - sema.availablePermits())
                            + "个线程并发");
                    try {
                        Thread.sleep(new Random().nextInt(5000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("线程" + Thread.currentThread().getName()
                            + "即将离开");
                    sema.release(); // 一个线程释放就会允许另外的线程进入
                }
            });
        }
        es.shutdown();
    }
}
运行结果:
线程pool-1-thread-2已进入,当前有2个线程并发
线程pool-1-thread-1已进入,当前有2个线程并发
线程pool-1-thread-3已进入,当前有3个线程并发
线程pool-1-thread-1即将离开
线程pool-1-thread-4已进入,当前有3个线程并发
线程pool-1-thread-4即将离开
线程pool-1-thread-5已进入,当前有3个线程并发
线程pool-1-thread-2即将离开
线程pool-1-thread-9已进入,当前有3个线程并发
线程pool-1-thread-5即将离开
线程pool-1-thread-7已进入,当前有3个线程并发
线程pool-1-thread-3即将离开
线程pool-1-thread-6已进入,当前有3个线程并发
线程pool-1-thread-6即将离开
线程pool-1-thread-8已进入,当前有3个线程并发
线程pool-1-thread-9即将离开
线程pool-1-thread-10已进入,当前有3个线程并发
线程pool-1-thread-7即将离开
线程pool-1-thread-8即将离开
线程pool-1-thread-10即将离开




喜欢0 评分0
游客

返回顶部