一、概念
Semaphore也是一个线程同步的辅助类,可以维护当前访问自身的线程个数,并提供了同步机制。使用Semaphore可以控制同时访问资源的线程个数,例如,实现一个文件允许的并发访问数。
二、Semaphore的主要方法:
void acquire():从此信号量获取一个许可,在提供一个许可前一直将线程阻塞,否则线程被中断。
void release():释放一个许可,将其返回给信号量。
int availablePermits():返回此信号量中当前可用的许可数。
boolean hasQueuedThreads():查询是否有线程正在等待获取。
三、为什么要用Semaphore
Semaphore有两个目的,第一个是多个共享资源互斥使用,第二个是并发线程数的控制,具体例子如下:
6个线程同时执行一个工作,只允许最多3个同时执行,由上图可以看出,release() 释放资源的顺序不一定遵循进入的顺序。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class SemaphoreTest {
public static void main(String[] args) {
testSemaphore();
}
public static void testSemaphore() {
ExecutorService executorService = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(200);
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
//请求获得许可,如果有可获得的许可则继续往下执行,许可数减1。否则进入阻塞状态
semaphore.acquire();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程" + Thread.currentThread().getName() +
"进入,当前已有" + (3 - semaphore.availablePermits()) + "个并发");
try {
Thread.sleep((long) (Math.random() * 10000));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程" + Thread.currentThread().getName() +
"即将离开");
semaphore.release();//释放许可,许可数加1
//下面代码有时候执行不准确,因为其没有和上面的代码合成原子单元
System.out.println("线程" + Thread.currentThread().getName() +
"已离开,当前已有" + (3 - semaphore.availablePermits()) + "个并发");
}
};
executorService.execute(runnable);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
三、 同ReentrantLock比较,Semaphore内部也是依靠一个继承自AbstractQueuedSynchronizer的Sync抽象类型的类成员变量sync来实现主要功能的, 单个信号量的Semaphore对象可以实现互斥锁的功能,并且可以是由一个线程获得了“锁”,再由另一个线程释放“锁”,这可应用于死锁恢复的一些场合。
import java.sql.BatchUpdateException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class LockTest {
public static void main(String[] args) {
final Business business = new Business();
ExecutorService executorService = Executors.newCachedThreadPool();
executorService.execute(new Runnable() {
@Override
public void run() {
business.service();
}
});
}
private static 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(); //释放锁
}
}
}
}
网友评论