使用java.util.concurrent包处理多线程

出处：http://blog.csdn.net/hjl_168562/article/details/8158023

1、使用拥有固定的线程数的线程池执行线程任务

package com.justin.thread.concurrent;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class TestThreadPool {

 public static void main(String args[]) throws InterruptedException {

  // only two threads

  ExecutorService exec = Executors.newFixedThreadPool(5);

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

   Runnable run = new Runnable() {

    public void run() {

     long time = (long) (Math.random() * 1000);

     System.out.println(Thread.currentThread().getName() + ":Sleeping " + time + "ms");

     try {

      Thread.sleep(time);

     } catch (InterruptedException e) {

     }

    }

   };

   exec.execute(run);

  }

  // must shutdown

  exec.shutdown();

 }

}

2、执行定期任务

public class TestScheduledThread {

 public static void main(String[] args) {

  final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

  final Runnable beeper = new Runnable() {

   int count = 0;

   public void run() {

    System.out.println(new Date() + " beep " + (++count));

   }

  };

  // 1秒钟后运行，并每隔2秒运行一次

  final ScheduledFuture<?> beeperHandle = scheduler.scheduleAtFixedRate(beeper, 1, 2, TimeUnit.SECONDS);

  // 2秒钟后运行，并每次在上次任务运行完后等待5秒后重新运行

  final ScheduledFuture<?> beeperHandle2 = scheduler.scheduleWithFixedDelay(beeper, 2, 5, TimeUnit.SECONDS);

  // 30秒后结束关闭任务，并且关闭Scheduler

  scheduler.schedule(new Runnable() {

   public void run() {

    beeperHandle.cancel(true);

    beeperHandle2.cancel(true);

    scheduler.shutdown();

   }

  }, 30, TimeUnit.SECONDS);

 }

}

3、多线程工程以完成同一件事情，而且在完成过程中，往往会等待其他线程都完成某一阶段后再执行，等所有线程都到达某一个阶段后再统一执行(比如有几个旅行团需要途经深圳、广州、韶关、长沙最后到达武汉。旅行团中有自驾游的，有徒步的，有乘坐旅游大巴的；这些旅行团同时出发，并且每到一个目的地，都要等待其他旅行团到达此地后再同时出发，直到都到达终点站武汉)

package com.justin.thread.concurrent;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class TestCyclicBarrier {

 // 徒步需要的时间: Shenzhen, Guangzhou, Shaoguan, Changsha, Wuhan

 private static int[] timeWalk = { 5, 8, 15, 15, 10 };

 // 自驾游

 private static int[] timeSelf = { 1, 3, 4, 4, 5 };

 // 旅游大巴

 private static int[] timeBus = { 2, 4, 6, 6, 7 };

 static String now() {

  SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

  return sdf.format(new Date()) + ": ";

 }

 static class Tour implements Runnable {

  private int[] times;

  private CyclicBarrier barrier;

  private String tourName;

  public Tour(CyclicBarrier barrier, String tourName, int[] times) {

   this.times = times;

   this.tourName = tourName;

   this.barrier = barrier;

  }

  public void run() {

   try {

    Thread.sleep(times[0] * 1000);

    System.out.println(now() + tourName + " Reached Shenzhen");

    barrier.await();

    Thread.sleep(times[1] * 1000);

    System.out.println(now() + tourName + " Reached Guangzhou");

    barrier.await();

    Thread.sleep(times[2] * 1000);

    System.out.println(now() + tourName + " Reached Shaoguan");

    barrier.await();

    Thread.sleep(times[3] * 1000);

    System.out.println(now() + tourName + " Reached Changsha");

    barrier.await();

    Thread.sleep(times[4] * 1000);

    System.out.println(now() + tourName + " Reached Wuhan");

    barrier.await();

   } catch (InterruptedException e) {

   } catch (BrokenBarrierException e) {

   }

  }

 }

 public static void main(String[] args) {

  // 三个旅行团

  CyclicBarrier barrier = new CyclicBarrier(3);

  ExecutorService exec = Executors.newFixedThreadPool(3);

  exec.submit(new Tour(barrier, "WalkTour", timeWalk));

  exec.submit(new Tour(barrier, "SelfTour", timeSelf));

  exec.submit(new Tour(barrier, "BusTour", timeBus));

  exec.shutdown();

 }

}

4、BlockingQueue,该类主要提供了两个方法put()和take()，前者将一个对象放到队列中，如果队列已经满了，就等待直到有空闲节点；后者从head取一个对象，如果没有对象，就等待直到有可取的对象

package com.justin.thread.concurrent;

import java.io.File;

import java.io.FileFilter;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.atomic.AtomicInteger;

public class TestBlockingQueue {

 static long randomTime() {

  return (long) (Math.random() * 1000);

 }

 public static void main(String[] args) {

  // 能容纳100个文件

  final BlockingQueue queue = new LinkedBlockingQueue(100);

  // 线程池

  final ExecutorService exec = Executors.newFixedThreadPool(5);

  final File root = new File("D:\\Workspace\\Study");

  // 完成标志

  final File exitFile = new File("");

  // 读个数

  final AtomicInteger rc = new AtomicInteger();

  // 写个数

  final AtomicInteger wc = new AtomicInteger();

  // 读线程

  Runnable read = new Runnable() {

   public void run() {

    scanFile(root);

    scanFile(exitFile);

   }

   public void scanFile(File file) {

    if (file.isDirectory()) {

     File[] files = file.listFiles(new FileFilter() {

      public boolean accept(File pathname) {

       return pathname.isDirectory() || pathname.getPath().endsWith(".java");

      }

     });

     for (File one : files)

      scanFile(one);

    } else {

     try {

      int index = rc.incrementAndGet();

      System.out.println("Read0: " + index + " " + file.getPath());

      queue.put(file);

     } catch (InterruptedException e) {

     }

    }

   }

  };

  exec.submit(read);

  // 四个写线程

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

   // write thread

   final int NO = index;

   Runnable write = new Runnable() {

    String threadName = "Write" + NO;

    public void run() {

     while (true) {

      try {

       Thread.sleep(randomTime());

       int index = wc.incrementAndGet();

       File file = (File) queue.take();

       // 队列已经无对象

       if (file == exitFile) {

        // 再次添加"标志"，以让其他线程正常退出

        queue.put(exitFile);

        break;

       }

       System.out.println(threadName + ": " + index + " " + file.getPath());

      } catch (InterruptedException e) {

      }

     }

    }

   };

   exec.submit(write);

  }

  exec.shutdown();

 }

}

5、CountDownLatch,下面的例子简单的说明了CountDownLatch的使用方法，模拟了100米赛跑，10名选手已经准备就绪，只等裁判一声令下。当所有人都到达终点时，比赛结束

package com.justin.thread.concurrent;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class TestCountDownLatch {

 public static void main(String[] args) throws InterruptedException {

  // 开始的倒数锁

  final CountDownLatch begin = new CountDownLatch(1);

  // 结束的倒数锁

  final CountDownLatch end = new CountDownLatch(10);

  // 十名选手

  final ExecutorService exec = Executors.newFixedThreadPool(10);

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

   final int NO = index + 1;

   Runnable run = new Runnable() {

    public void run() {

     try {

      begin.await();

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

      System.out.println("No." + NO + " arrived");

     } catch (InterruptedException e) {

     } finally {

      end.countDown();

     }

    }

   };

   exec.submit(run);

  }

  System.out.println("Game Start");

  begin.countDown();

  end.await();

  System.out.println("Game Over");

  exec.shutdown();

 }

}

6、Future,比如用网页浏览器浏览新闻时，最重要的是要显示文字内容，至于与新闻相匹配的图片就没有那么重要的，所以此时首先保证文字信息先显示，而图片信息会后显示，但又不能不显示，由于下载图片是一个耗时的操作，所以必须一开始就得下载

package com.justin.thread.concurrent;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class TestFutureTask {

 public static void main(String[] args) throws InterruptedException, ExecutionException {

  final ExecutorService exec = Executors.newFixedThreadPool(5);

  Callable call = new Callable() {

   public String call() throws Exception {

    Thread.sleep(1000 * 5);

    return "Other less important but longtime things.";

   }

  };

  Future task = exec.submit(call);

  // 重要的事情

  Thread.sleep(1000 * 3);

  System.out.println("Let’s do important things.");

  // 其他不重要的事情

  String obj = (String) task.get();

  System.out.println(obj);

  // 关闭线程池

  exec.shutdown();

 }

}

7、ExecutorCompletionService, 考虑以下场景：浏览网页时，浏览器了5个线程下载网页中的图片文件，由于图片大小、网站访问速度等诸多因素的影响，完成图片下载的时间就会有很大的不同。如果先下载完成的图片就会被先显示到界面上，反之，后下载的图片就后显示。

Java的并发库的CompletionService可以满足这种场景要求。该接口有两个重要方法：submit()和take()。submit用于提交一个runnable或者callable，一般会提交给一个线程池处理；而take就是取出已经执行完毕runnable或者callable实例的Future对象，如果没有满足要求的，就等待了。 CompletionService还有一个对应的方法poll，该方法与take类似，只是不会等待，如果没有满足要求，就返回null对象

package com.justin.thread.concurrent;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorCompletionService;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class CompletionService {

 public static void main(String[] args) throws InterruptedException, ExecutionException {

  ExecutorService exec = Executors.newFixedThreadPool(10);

  ExecutorCompletionService serv = new ExecutorCompletionService(exec);

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

   final int NO = index;

   Callable downImg = new Callable() {

    public String call() throws Exception {

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

     return "Downloaded Image " + NO;

    }

   };

   serv.submit(downImg);

  }

  Thread.sleep(1000 * 2);

  System.out.println("Show web content");

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

   Future task = serv.take();

   String img = (String) task.get();

   System.out.println(img);

  }

  System.out.println("End");

  // 关闭线程池

  exec.shutdown();

 }

}

8、Semaphore,下面的Demo中申明了一个只有5个许可的Semaphore，而有20个线程要访问这个资源，通过acquire()和release()获取和释放访问许可

package com.justin.thread.concurrent;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Semaphore;

public class TestSemaphore {

 public static void main(String[] args) {

  // 线程池

  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 {

      // 获取许可

      semp.acquire();

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

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

      // 访问完后，释放

      semp.release();

     } catch (InterruptedException e) {

     }

    }

   };

   exec.execute(run);

  }

  // 退出线程池

  exec.shutdown();

 }

}