疯狂java


您现在的位置: 疯狂软件 >> 新闻资讯 >> 正文

Java线程并发控制基础知识


 

  线程池、线程同步、互斥锁、读写锁、原子数、唤醒、通知、信号量、线程交换队列

  线程池

  推荐用ThreadPoolExecutor的工厂构造类Executors来管理线程池,线程复用线程池开销较每次申请新线程小,具体看代码以及注释

  public class TestThread {

  /**

  * 使用线程池的方式是复用线程的(推荐)

  * 而不使用线程池的方式是每次都要创建线程

  * Executors.newCachedThreadPool(),该方法返回的线程池是没有线程上限的,可能会导致过多的内存占用

  * 建议使用Executors.newFixedThreadPool(n)

  *

  * 有兴趣还可以看下定时线程池:SecheduledThreadPoolExecutor

  */

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

  int nThreads = 5;

  /**

  * Executors是ThreadPoolExecutor的工厂构造方法

  */

  ExecutorService executor = Executors.newFixedThreadPool(nThreads);

  //submit有返回值,而execute没有返回值,有返回值方便Exception的处理

  Future res = executor.submit(new ConsumerThread());

  //executor.execute(new ConsumerThread());

  /**

  * shutdown调用后,不可以再submit新的task,已经submit的将继续执行

  * shutdownNow试图停止当前正执行的task,并返回尚未执行的task的list

  */

  executor.shutdown();

  //配合shutdown使用,shutdown之后等待所有的已提交线程运行完,或者到超时。继续执行后续代码

  executor.awaitTermination(1, TimeUnit.DAYS);

  //打印执行结果,出错的话会抛出异常,如果是调用execute执行线程那异常会直接抛出,不好控制,submit提交线程,调用res.get()时才会抛出异常,方便控制异常

  System.out.println("future result:"+res.get());

  }

  static class ConsumerThread implements Runnable{

  @Override

  public void run() {

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

  System.out.println(i);

  }

  }

  }

  }

  输出:

  0

  1

  2

  3

  4

  future result:null

  线程同步

  synchronized(this)和synchronized(MyClass.class)区别:前者与加synchronized的成员方法互斥,后者和加synchronized的静态方法互斥

  synchronized的一个应用场景是单例模式的,双重检查锁

  public class Singleton {

  private volatile static Singleton singleton;

  private Singleton (){}

  public static Singleton getSingleton() {

  if (singleton == null) {

  synchronized (Singleton.class) {

  if (singleton == null) {

  singleton = new Singleton();

  }

  }

  }

  return singleton;

  }

  }

  注意:不过双重检查锁返回的实例可能是没有构造完全的对象,高并发的时候直接使用有问题,不知道在新版的java里是否解决了

  所以有了内部类方式的单例模式,这样的单例模式有了延迟加载的功能(还有一种枚举方式的单例模式,用的不多,有兴趣的可以上网查)

  //(推荐)延迟加载的单例模式

  public class Singleton {

  private static class SingletonHolder {

  private static final Singleton INSTANCE = new Singleton();

  }

  private Singleton (){}

  public static final Singleton getInstance() {

  return SingletonHolder.INSTANCE;

  }

  }

  若不要延迟加载,在类加载的时候实例化对象,那直接这么写,如下:

  public class Singleton {

  private static Singleton instance = new Singleton();

  private Singleton (){}

  public static Singleton getInstance() {

  return instance;

  }

  }

  volatile保证同一变量在多线程中的可见性,所以它更多是用于修饰作为开关状态的变量

  用synchronized修饰变量的get和set方法,不但可以保证和volatile修饰变量一样的效果(获取最新值),因为synchronized不仅会把当前线程修改的变量的本地副本同步给主存,还会从主存中读取数据更新本地副本。而且synchronized还有互斥的效果,可以有效控制并发修改一个值,因为synchronized保证代码块的串行执行。如果只要求获取最新值的特性,用volatile就好,因为volatile比较轻量,性能较好。

  互斥锁、读写锁

  ReentrantLock 和 ReentrantReadWriteLock

  JDK5增加了ReentrantLock这个类因为两点:

  1.ReentrantLock提供了tryLock方法,tryLock调用的时候,如果锁被其他线程(同一个线程两次调用tryLock也都返回true)持有,那么tryLock会立即返回,返回结果是false.lock()方法会阻塞。

  2.构造RenntrantLock对象可以接收一个boolean类型的参数,描述锁公平与否的函数。公平锁的好处是等待锁的线程不会饿死,但是整体效率相对低一些;非公平锁的好处是整体效率相对高一些。

  注意:使用ReentrantLock后,需要显式地进行unlock,所以建议在finally块中释放锁,如下:

  lock.lock();

  try {

  //do something

  }

  finally {

  lock.unlock();

  }

  ReentrantReadWriteLock与ReentrantLock的用法类似,差异是前者原子数

  除了用互斥锁控制变量的并发修改之外,jdk5中还增加了原子类,通过比较并交换(硬件CAS指令)来避免线程互斥等待的开销,进而完成超轻量级的并发控制,一般用来高效的获取递增计数器。

  AtomicInteger counter = new AtomicInteger();

  counter.incrementAndGet();

  counter.decrementAndGet();

  可以简单的理解为以下代码,增加之后与原先值比较,如果发现增长不一致则循环这个过程。代码如下

  public class CasCounter {

  private SimulatedCAS value;

  public int getValue() {

  return value.getValue();

  }

  public int increment() {

  int oldValue = value.getValue();

  while (value.compareAndSwap(oldValue, oldValue + 1) != oldValue)

  oldValue = value.getValue();

  return oldValue + 1;

  }

  }

  可以看IBM工程师的一篇文章 Java 理论与实践: 流行的原子

  唤醒、通知

  wait,notify,notifyAll是java的Object对象上的三个方法,多线程中可以用这些方法完成线程间的状态通知。

  notify是唤醒一个等待线程,notifyAll会唤醒所有等待线程。

  CountDownLatch主要提供的机制是当多个(具体数量等于初始化CountDownLatch时的count参数的值)线程都到达了预期状态或完成预期工作时触发事件,其他线程可以等待这个事件来触发后续工作。

  举个例子,大数据分拆给多个线程进行排序,比如主线程

  CountDownLatch latch = new CountDownLatch(5);

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

  threadPool.execute(new MyRunnable(latch,datas));

  }

  latch.await();

  //do something 合并数据

  MyRunnable的实现代码如下

  public void run() {

  //do something数据排序

  latch.countDown();

  //继续自己线程的工作,与CyclicBarrier最大的不同,稍后马上讲

  }

  CyclicBarrier循环屏障,协同多个线程,让多个线程在这个屏障前等待,直到所有线程都到达了这个屏障时,再一起继续执行后面的动作。

  使用CyclicBarrier可以重写上面的排序代码

  主线程如下

  CyclicBarrier barrier = new CyclicBarrier(5+1); //主线程也要消耗一个await,所以+1

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

  threadPool.execute(new MyRunnable(barrier,datas));//如果线程池线程数过少,就会发生死锁

  }

  barrier.await();

  //合并数据

  MyRunnable代码如下

  public void run() {

  //数据排序

  barrier.await();

  }

  //全部 count+1 await之后(包括主线程),之后的代码才会一起执行信号量

  Semaphore用于管理信号量,与锁的最大区别是,可以通过令牌的数量,控制并发数量,当管理的信号量只有1个时,就退化到互斥锁。

  例如我们需要控制远程方法的并发量,代码如下

  semaphore.acquire(count);

  try {

  //调用远程方法

  }

  finally {

  semaphore.release(count);

  }

  线程交换队列

  Exchanger用于在两个线程之间进行数据交换,线程会阻塞在Exchanger的exchange方法上,直到另外一个线程也到了同一个Exchanger的exchanger方法时,二者进行交换,然后两个线程继续执行自身相关代码。

  public class TestExchanger {

  static Exchanger exchanger = new Exchanger();

  public static void main(String[] args) {

  new Thread() {

  public void run() {

  int a = 1;

  try {

  a = (int) exchanger.exchange(a);

  } catch (Exception e) {

  e.printStackTrace();

  }

  System.out.println("Thread1: "+a);

  }

  }.start();

  new Thread() {

  public void run() {

  int a = 2;

  try {

  a = (int) exchanger.exchange(a);

  } catch (Exception e) {

  e.printStackTrace();

  }

  System.out.println("Thread2: "+a);

  }

  }.start();

  }

  }

  输出结果:

  Thread2: 1

  Thread1: 2

  并发容器

  CopyOnWrite思路是在更改容器时,把容器写一份进行修改,保证正在读的线程不受影响,适合应用在读多写少的场景,因为写的时候重建一次容器。

  以Concurrent开头的容器尽量保证读不加锁,并且修改时不影响读,所以会达到比使用读写锁更高的并发性能通过readLock()和writeLock()两个方法获得相关的读锁和写锁操作。