使用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();
}
}

使用java.util.concurrent包处理多线程相关推荐

java.util.concurrent包API学习笔记
newFixedThreadPool 创建一个固定大小的线程池. shutdown():用于关闭启动线程,如果不调用该语句,jvm不会关闭. awaitTermination():用于等待子线程结束, ...
【ArrayList】为什么java.util.concurrent 包里没有并发的ArrayList实现？
2019独角兽企业重金招聘Python工程师标准>>> 为什么java.util.concurrent 包里没有并发的ArrayList实现? 问:JDK 5在java.util.c ...
高并发编程基础（java.util.concurrent包常见类基础）
JDK5中添加了新的java.util.concurrent包,相对同步容器而言,并发容器通过一些机制改进了并发性能.因为同步容器将所有对容器状态的访问都串行化了,这样保证了线程的安全性,所以这种方法 ...
java.util.concurrent包
本文是我们学院课程中名为Java Concurrency Essentials的一部分 . 在本课程中,您将深入探讨并发的魔力. 将向您介绍并发和并发代码的基础知识,并学习诸如原子性,同步和线程安全之 ...
Java高并发编程学习（三）java.util.concurrent包
简介我们已经学习了形成Java并发程序设计基础的底层构建块,但对于实际编程来说,应该尽可能远离底层结构.使用由并发处理的专业人士实现的较高层次的结构要方便得多.要安全得多.例如,对于许多线程问题,可 ...
java.util.concurrent 包下面的所有类
java.util.concurrent 包下面的所有类原子操作数类: java.util.concurrent.atomic.AtomicBoolean.class java.util.concu ...
java.util.concurrent包详细分析--转
原文地址:http://blog.csdn.net/windsunmoon/article/details/36903901 概述 Java.util.concurrent 包含许多线程安全.测试良好 ...
java concurrent 框架,java.util.concurrent 包下的 Synchronizer 框架
看完书 java concurrency in practice 当然是想找点啥好玩的东东玩玩. 当看到了Doug Lee 的论文 << The java.util.concurrent ...
jdk8中java.util.concurrent包分析
并发框架分类 1. Executor相关类 Interfaces. Executor is a simple standardized interface for defining custom th ...

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

使用java.util.concurrent包处理多线程相关推荐

最新文章

热门文章