Java分布式应用学习笔记05多线程下的并发同步器----前篇

1.  前言

JDK提供的并发包，除了上一篇提到的用于集合外，还有线程的调度、协作、调度等等功能。上篇提到过，线程之间除了竞争关系，还有协作关系。在高并发环境下有效利用Java并发包解决线程之间协作的特殊场景。在并行计算，尤其是多线程计算的结果集合并的时候都需要用到这些并发同步器。还有一种使用场景，就是跨越多台机器（实机）的多线程进行并行运算，需要将多台机器进行结果集的汇总，合并。其原理核心也是使用这些并发协作包。

2.  FutureTask

FutureTask是进行并行结果集合并的类，此类是Future接口的实现。在主线程中启动多个线程进行并发计算，之后再根据各个线程的执行结果进行汇总，归并，得出一个总的结果，这个多线程可以是在一台机器上，充分利用多核CPU硬件，在科研单位可能分布式集群环境一起并发计算一个大任务，每个机器相当于一个线程，执行完毕后将反馈的结果返回来进行合并后才是最终的结果。而主线程可以等待分线程的结果，也可以不等待，全凭具体业务需要而定，不过一般情况下还是要等一等分线程的结果才能往下执行的。如果不等分线程，也可以在主线程中不再理会分线程即可。

举个实例，比如这时候东方不败要想练成《葵花宝典》，那么需要前提条件是2个，第一手中得有《葵花宝典》秘籍，第二就是挥刀自宫。恩，挥刀自宫这个主线程——东方不败可以自己完成，夺取《葵花宝典》可以派别人——兄弟童柏雄去干，2条线并行实施，等另一个人取得《葵花宝典》了，这边主线程也挥刀自宫了，行了，能练了！

咱先看代码行吧

package threadConcurrent.test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

/**
 * 分线程汇总
 * @author liuyan
 */
public class FutureTaskDemo {

	@SuppressWarnings("unchecked")
	public static void main(String[] args) {

		// 初始化一个Callable对象和FutureTask对象
		Callable otherPerson = new OtherPerson();

		// 由此任务去执行
		FutureTask futureTask = new FutureTask(otherPerson);

		// 使用futureTask创建一个线程
		Thread newhread = new Thread(futureTask);
		
		System.out.println("newhread线程现在开始启动，启动时间为：" + System.nanoTime()
				+ " 纳秒");
		
		newhread.start();
		
		System.out.println("主线程——东方不败，开始执行其他任务");
		
		System.out.println("东方不败开始准备小刀，消毒...");

		//兄弟线程在后台的计算线程是否完成，如果未完成则等待
		//阻塞
		while (!futureTask.isDone()) {
			
			try {
				Thread.sleep(500);
				System.out.println("东方不败：“等兄弟回来了，我就和小弟弟告别……颤抖……”");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		System.out.println("newhread线程执行完毕，此时时间为" + System.nanoTime());
		String result = null;
		try {
			result = (String) futureTask.get();

		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
		
		if("OtherPerson：：：经过一番厮杀取得《葵花宝典》".equals(result)){
			System.out.println("兄弟，干得好，我挥刀自宫了啊！");
		}else{
			System.out.println("还好我没自宫！否则白白牺牲了……");
		}
		
	}
}

@SuppressWarnings("all")
class OtherPerson implements Callable {

	@Override
	public Object call() throws Exception {

		// 先休息休息再拼命去！
		Thread.sleep(5000);
		String result = "OtherPerson：：：经过一番厮杀取得《葵花宝典》";
		System.out.println(result);
		return result;
	}

}

 在这个例子中主线程代表东方不败，分线程代表兄弟——童柏雄，主线程派出FutureTask，把它放置于一个线程对象中，之后线程开始启动，分支线程开始工作。主线程也没闲着，继续做自己的事情，消毒，做着心理斗争等等，通过一个阻塞的死循环，等待分线程的状态，调用分线程的futureTask.isDone()方法进行判断，看看兄弟是否执行结束了，结束了通过futureTask.get()将分线程的执行结果取出来，结果出来了，主线程根据分线程的执行结果再做决定。执行后，结果大家都明了。有一点需要说明的是，有可能分线程与主线程的执行不在一台物理机器上，分线程可以使用jms、webservic、rmi甚至socket技术请求远程的类为其服务。分线程根据远程返回的结果再返回给本机器的主线程，之后再做决策。分布式计算的核心原理也是如此，当然分布式计算比这个复杂得多，笔者只是说其核心的实现原理。

3.  Semaphore

Semaphore是限制多线程共享资源的一个东东，多线程并发访问一个资源的时候，可以限制线程最大使用个数，其他多出来的线程，没办法，耐心等着吧。这个例子在生活中比比皆是，在火车站售票处一共开设了5个窗口，也就表示在同一时间内，火车站的工作人员最多只能为5个人服务，那么高峰时其他人呢，理想的情况下是排队等着，不理想的情况下是，等待的队列没有秩序，有的只是拳头和权势，没有办法，人家的爸爸是李刚，撞人都没事何况是排队买票了，人家说的就是王法。当然了，这个咱们看具体程序。

package threadConcurrent.test;

import java.util.Random;
import java.util.concurrent.Semaphore;

/**
 * 使用Semaphore，限制可以执行的线程数，空闲资源放到队列中等待
 * 
 * @author liuyan
 */
public class SemaphoreDemo {

	public static void main(String[] args) {
		Runnable limitedCall = new Runnable() {

			// 随机生成数
			final Random rand = new Random();

			// 限制只有3个资源是活动的,第二个参数为true则是按照标准“队列”结构先进先出
			final Semaphore available = new Semaphore(5, true);
			int count = 0;

			public void run() {
				int time = rand.nextInt(10);
				int num = count++;

				try {

					// 请求资源
					available.acquire();

					int needTime = time * 2000;

					System.out.println("乘客" + num + "买票需要[ " + needTime
							+ " 秒]... #");

					Thread.sleep(needTime);

					System.out.println("乘客" + num + "买完了 # !");

					// 运行完了就释放
					available.release();
				} catch (InterruptedException intEx) {
					intEx.printStackTrace();
				}
			}
		};

		for (int i = 0; i < 25; i++)
			new Thread(limitedCall).start();
	}
}

 注释已经写得比较明确了，构建Semaphore的时候，第一个参数代表线程的执行的最大数目，第二个参数是按照队列的形式将未执行的线程放到队列中，当有线程执行完了后，按照先进先出的原则，进行线程的唤醒，执行。即便是main启动了25个线程，那么其余的线程要向执行也要等前面的线程执行完毕后才能有资格执行。要想让线程按规矩执行，首先应该先向资源池申请资源，available.acquire();就是请求资源池给个资源，如果资源池当前有空闲资源，那么线程就可以正常运行了，如果没有，没办法，排队吧啊。线程运行完毕了，要记得归还资源available.release();如果构造Semaphore的时候没指定第二个参数，或者第二个参数为false，估计您有幸能见到我之前说的李刚的儿子的现象！在此不再赘述。

4.  ScheduledFuture

提到Quartz，大家都知道他是一个负责任务调度的开源工具，使用它可以轻易地在某一时段，某一频率执行相关业务功能。如果仅仅是简单的根据某些时间频率执行某些任务，其实到不必屠龙刀杀小鸡，使用ScheduledFuture可以轻松解决此类频率的问题，启动另一个线程来，在某一个时间频率执行代码。这个还是举个例子吧，战争年代巡视城防，赵云带一个小兵去巡视城防，赵云是将军，每5秒钟巡视一次士兵，看看士兵有没有偷懒，士兵比较累，每1秒巡视一次城防，不能睡觉。如下程序

package threadConcurrent.test;

import static java.util.concurrent.TimeUnit.SECONDS;

import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;

/**
 * 时间频率调度
 * @author liuyan
 */
public class ScheduledFutureDemo {

	@SuppressWarnings("unchecked")
	public static void main(String[] args) {

		// 线程池开辟2个线程
		final ScheduledExecutorService scheduler = Executors
				.newScheduledThreadPool(2);

		// 将军
		final Runnable general = new Runnable() {
			int count = 0;

			public void run() {
				System.out.println(Thread.currentThread().getName() + ":"
						+ new Date() + "赵云巡视来了 " + (++count));
			}
		};
		
		// 士兵
		final Runnable soldier = new Runnable() {
			int count = 0;

			public void run() {
				System.out.println(Thread.currentThread().getName() + ":"
						+ new Date() + "士兵巡视来了 " + (++count));
			}
		};

		// 1秒钟后运行，并每隔2秒运行一次
		final ScheduledFuture beeperHandle1 = scheduler.scheduleAtFixedRate(
				soldier, 1, 1, SECONDS);

		// 5秒钟后运行，并每隔2秒运行一次
		final ScheduledFuture beeperHandle2 = scheduler.scheduleWithFixedDelay(
				general, 5, 5, SECONDS);

		// 30秒后结束关闭任务，并且关闭Scheduler
		scheduler.schedule(new Runnable() {
			public void run() {
				beeperHandle1.cancel(true);
				beeperHandle2.cancel(true);
				scheduler.shutdown();
			}
		}, 60, SECONDS);
	}

}

 

程序的注释已经明白了，在此不再赘述。

评论

30 楼 OyazO 2013-03-21  

final Random rand = new Random(); 
 
            // 限制只有3个资源是活动的,第二个参数为true则是按照标准“队列”结构先进先出 
            final Semaphore available = new Semaphore(5, true); 
           int count = 0; 
 
            public void run() { 
                int time = rand.nextInt(10); 
                int num = count++; 
 
                try { 
 
                    // 请求资源 
                    available.acquire(); 
 
                    int needTime = time * 2000; 
 
                    System.out.println("乘客" + num + "买票需要[ " + needTime 
                            + " 秒]... #"); 
 
                    Thread.sleep(needTime); 
 
                    System.out.println("乘客" + num + "买完了 # !"); 
 
                    // 运行完了就释放 
                    available.release(); 
                } catch (InterruptedException intEx) { 
                    intEx.printStackTrace(); 
                } 
            } 
        }; 


count 为共享资源 ？？这样安全吗？？

29 楼 fantastie1209 2011-08-17  

28 楼 myeclipse8868 2011-08-15  

持续期待中……

27 楼 suhuanzheng7784877 2011-08-15  

主食是有问题，是笔者代码写完后有设置了几个参数试了试，结果注释忘了改了。谢谢各位仁兄的指正

26 楼 suhuanzheng7784877 2011-08-15  

引用

你要是愤青能愤到这么深刻的，你也NB

兄弟，一篇博客而已，没必要……用天津话讲：“乐呵乐呵得了”

25 楼 suhuanzheng7784877 2011-08-15  

你若无情我便休 写道

酣畅淋漓，一气呵成。 确实很暴力

过誉了

24 楼 suhuanzheng7784877 2011-08-15  

cn-done 写道

楼主的代码注释对于参数的说明，略微有点问题。
eg:
 

// 1秒钟后运行，并每隔2秒运行一次  
        final ScheduledFuture beeperHandle1 = scheduler.scheduleAtFixedRate(  
                soldier, 1, 1, SECONDS);  
  
        // 5秒钟后运行，并每隔2秒运行一次  
        final ScheduledFuture beeperHandle2 = scheduler.scheduleWithFixedDelay(  
                general, 5, 5, SECONDS); 

但是瑕不掩瑜！希望LZ分享出更多的好作品

哦，是~~~是代码写完后有设置了几个参数试了试，结果注释忘了改了。谢谢各位仁兄的指正

23 楼 suhuanzheng7784877 2011-08-15  

zhaoxjmail 写道

// 限制只有3个资源是活动的,第二个参数为true则是按照标准“队列”结构先进先出  
  final Semaphore available = new Semaphore(5, true); 



以上注释有问题吧?

哎呀~是啊~~~注释有问题

22 楼 suhuanzheng7784877 2011-08-15  

stan001140 写道

半天还没讲啊!

？？？？？

21 楼 stan001140 2011-08-14  

半天还没讲啊!

20 楼 你若无情我便休 2011-08-14  

酣畅淋漓，一气呵成。 确实很暴力

19 楼 zhaoxjmail 2011-08-14  

// 限制只有3个资源是活动的,第二个参数为true则是按照标准“队列”结构先进先出  
  final Semaphore available = new Semaphore(5, true); 



以上注释有问题吧?

18 楼 cn-done 2011-08-14  

楼主的代码注释对于参数的说明，略微有点问题。
eg:
 

// 1秒钟后运行，并每隔2秒运行一次  
        final ScheduledFuture beeperHandle1 = scheduler.scheduleAtFixedRate(  
                soldier, 1, 1, SECONDS);  
  
        // 5秒钟后运行，并每隔2秒运行一次  
        final ScheduledFuture beeperHandle2 = scheduler.scheduleWithFixedDelay(  
                general, 5, 5, SECONDS); 

但是瑕不掩瑜！希望LZ分享出更多的好作品

17 楼 qijinha_toni 2011-08-13  

引用

可以看出楼主骨子里是个愤青

你要是愤青能愤到这么深刻的，你也NB

16 楼 thomas_mule0086 2011-08-13  

引用

唉~又让我深深的惭愧了。。。。

lz才让我深深的惭愧

15 楼 hufei023090 2011-08-12  

原来jdk还有这功能，还真不知道，谢谢lz

14 楼 suhuanzheng7784877 2011-08-12  

jiang_yun198701 写道

引用

理想的情况下是排队等着，不理想的情况下是，等待的队列没有秩序，有的只是拳头和权势，没有办法，人家的爸爸是李刚，撞人都没事何况是排队买票了，人家说的就是王法。

可以看出楼主骨子里是个愤青

呵呵，这和愤青貌似没关系吧。我个人感觉啊，其实很多技术上的东西，都可以用现实实例来说事儿的。真的。

13 楼 jiang_yun198701 2011-08-12  

引用

理想的情况下是排队等着，不理想的情况下是，等待的队列没有秩序，有的只是拳头和权势，没有办法，人家的爸爸是李刚，撞人都没事何况是排队买票了，人家说的就是王法。

可以看出楼主骨子里是个愤青

12 楼 qianjingtu1008 2011-08-12  

lvjun106 写道

很黄很暴力~~~

人家lz是比喻，你看看lz之前的文章就知道了。反正我们这边的圈子都挺喜欢lz的系列的。学东西之余还能给我们带来清爽一笑。哈哈

11 楼 suhuanzheng7784877 2011-08-12  

yuediaoyuan0809 写道

lz的文章风格有一种朴实感。我之前没接触过你说的这个东西，看了lz的文章之后就觉得明白了，并且知道在什么情况下用了。比那些故弄玄虚还装X的人强多了，希望你继续保持这种风格。

唉~又让我深深的惭愧了。。。。