线程同步工具

发表于 | 分类于

Semaphere同步工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package cn.itcast.heima2;

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 service = Executors.newCachedThreadPool();
		final  Semaphore sp = new Semaphore(3);
		for(int i=0;i<10;i++){
			Runnable runnable = new Runnable(){
					public void run(){
					try {
						sp.acquire();
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
					System.out.println("线程" + Thread.currentThread().getName() +
							"进入,当前已有" + (3-sp.availablePermits()) + "进入");
					try {
						Thread.sleep((long)(Math.random()*10000));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("线程" + Thread.currentThread().getName() +
							"即将离开");
					sp.release();
					//todo 
					System.out.println("线程" + Thread.currentThread().getName() +
							"已离开,当前已有" + (3-sp.availablePermits()) + "进入");
				}
			};
			service.execute(runnable);			
		}
	}

}

无论是内部锁synchronized还是重入锁ReentrantLock，一次都只允许一个线程访问某一个资源，而信号量却可以指定多个线程同时访问某一资源；主要用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。
线程池里面有20个可重复使用的线程数量，但是信号量只有5个，也就是每次只能并发5个线程执行，其他线程阻塞。

信号量为5，可以认为线程池里有5把锁，每个线程调用acquire和release分别表示获取锁和释放锁，这样，通过信号量就可以调度多个线程的执行。
Semaphore保证了,我管理的那部分代码同一时刻最多可以有n个线程访问
Semaphore可以维护当前访问自身的线程个数，并提 供了同步机制。使用Semaphore可以控制同时访问资源的线程个数，例如，实现一个文件允许的并发访问数。Semaphore实现的功能就类似厕所有5个坑，假如有十个人要上厕所，那么同时能有多少个人去上厕所呢？同时只能有5 个人能够占用，当5个人中的任何一个人让开后，其中在等待的另外5个人中又有一个可以占用了。 另外等待的5个人中可以是随机获得优先机会，也可以是按照先来后到的顺序获得机会，这取决于构造Semaphore对象时 传入的参数选项。
单个信号量的Semaphore对象可以实现互斥锁的功能， 并且可以是由一个线程获得了“锁”，再由另一个线程释放“锁”，这可应用于死锁恢复的一些场合。

CyclicBarrier同步工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package cn.itcast.heima2;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CyclicBarrierTest {

	public static void main(String[] args) {
		ExecutorService service = Executors.newCachedThreadPool();
		final  CyclicBarrier cb = new CyclicBarrier(3);//todo
		for(int i=0;i<3;i++){
			Runnable runnable = new Runnable(){
					public void run(){
					try {
						Thread.sleep((long)(Math.random()*10000));	
						System.out.println("线程" + Thread.currentThread().getName() +
								"即将到达集合地点1,当前已有" + (cb.getNumberWaiting()+1) + "已经到达," + (cb.getNumberWaiting()==2?"都到齐了,继续走啊":"正在等候"));
						cb.await();//todo 每次都等待,如果其他也全部到达,此时cb为3,则所有线程进行下一步
						
						Thread.sleep((long)(Math.random()*10000));
						System.out.println("线程" + Thread.currentThread().getName() +
								"即将到达集合地点2,当前已有" + (cb.getNumberWaiting()+1) + "已经到达," + (cb.getNumberWaiting()==2?"都到齐了,继续走啊":"正在等候"));
						cb.await();	
						Thread.sleep((long)(Math.random()*10000));
						System.out.println("线程" + Thread.currentThread().getName() +
								"即将到达集合地点3,当前已有" + (cb.getNumberWaiting()+1) + "已经到达," + (cb.getNumberWaiting()==2?"都到齐了,继续走啊":"正在等候"));
						cb.await();						
					} catch (Exception e) {
						e.printStackTrace();
					}				
				}
			};
			service.execute(runnable);
		}
		service.shutdown();
	}
}

字面意思循环屏障，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。我们暂且把这个状态就叫做barrier，当调用await()方法之后，线程就处于barrier了。

另外还有一点就是CyclicBarrier的计数器可以重置，例如设置的是拦截线程数量为2，但是有3个线程调用了await()方法表示到达了屏障，这个时候会出现最先达到屏障的两个线程顺利执行完毕，而最后到达的第三个线程则一直被阻塞，因为它等不到另外一个线程到达屏障了。

而如果拦截线程的数量依旧为2，但是有4个线程调用了await()方法，那么这4个线程是分两批执行的，前两个线程满足拦截的线程数，到达屏障后放行；然后CyclicBarrier的计数器重置，后面两个线程到达屏障后放行。

CountDownLatch同步工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
package cn.itcast.heima2;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CountdownLatchTest {

	public static void main(String[] args) {
		ExecutorService service = Executors.newCachedThreadPool();
		final CountDownLatch cdOrder = new CountDownLatch(1);
		final CountDownLatch cdAnswer = new CountDownLatch(3);		
		for(int i=0;i<3;i++){
			Runnable runnable = new Runnable(){
					public void run(){
					try {
						System.out.println("线程" + Thread.currentThread().getName() +
								"正准备接受命令");
						cdOrder.await();
						System.out.println("线程" + Thread.currentThread().getName() +
						"已接受命令");
						Thread.sleep((long)(Math.random()*10000));	
						System.out.println("线程" + Thread.currentThread().getName() +
								"回应命令处理结果");
						cdAnswer.countDown();						
					} catch (Exception e) {
						e.printStackTrace();
					}				
				}
			};
			service.execute(runnable);
		}		
		try {
			Thread.sleep((long)(Math.random()*10000));
		
			System.out.println("线程" + Thread.currentThread().getName() +
					"即将发布命令");
			cdOrder.countDown();
			System.out.println("线程" + Thread.currentThread().getName() +
			"已发布命令,正在等待结果");
			cdAnswer.await();
			System.out.println("线程" + Thread.currentThread().getName() +
			"已收到所有响应结果");
		} catch (Exception e) {
			e.printStackTrace();
		}				
		service.shutdown();

	}
}

利用它可以实现类似计数器的功能。比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。
这个CountDownLatch的构造函数接收一个int类型的参数作为计数器，N表示阻塞的线程必须等待N次countDown才能执行。

每次调用CountDownLatch的countDown方法时，N就会减1，而这个方法可以使用在任何地方，这里的N点可以是N个线程，也可以是一个线程中N个步骤。

而CountDownLatch的await方法则会阻塞当前线程，直到N为0的时候才能执行。

Exchanger同步工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
package cn.itcast.heima2;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExchangerTest {

	public static void main(String[] args) {
		ExecutorService service = Executors.newCachedThreadPool();
		final Exchanger exchanger = new Exchanger();
		service.execute(new Runnable(){
			public void run() {
				try {				

					String data1 = "zxx";
					System.out.println("线程" + Thread.currentThread().getName() +
					"正在准备把" + data1 +"换出去");
					Thread.sleep((long)(Math.random()*10000));
					String data2 = (String)exchanger.exchange(data1);
					System.out.println("线程" + Thread.currentThread().getName() +
					"换回的数据为" + data2);
				}catch(Exception e){
					
				}
			}	
		});
		service.execute(new Runnable(){
			public void run() {
				try {				

					String data1 = "lhm";
					System.out.println("线程" + Thread.currentThread().getName() +
					"正在准备把" + data1 +"换出去");
					Thread.sleep((long)(Math.random()*10000));					
					String data2 = (String)exchanger.exchange(data1);
					System.out.println("线程" + Thread.currentThread().getName() +
					"换回的数据为" + data2);
				}catch(Exception e){
					
				}				
			}	
		});		
	}
}

参考：
https://www.cnblogs.com/itermis/p/9025148.html
https://www.cnblogs.com/dongguacai/p/6023028.html