5W字高质量java并发系列详解教程(上)-附PDF下载

心已赠人 2022-09-11 01:29 79阅读 0赞

### 文章目录 ###

*  第一章 java.util.concurrent简介
 *   *  主要的组件
     *  Executor
     *  ExecutorService
     *  ScheduledExecutorService
     *  Future
     *  CountDownLatch
     *  CyclicBarrier
     *  Semaphore
     *  ThreadFactory
 *  第二章 java并发中的Synchronized关键词
 *   *  为什么要同步
     *  Synchronized关键词
     *   *  Synchronized Instance Methods
         *  Synchronized Static Methods
         *  Synchronized Blocks
 *  第三章 java中的Volatile关键字使用
 *   *  什么时候使用volatile
     *  Happens-Before
 *  第四章 wait和sleep的区别
 *   *  Wait和sleep的区别
     *  唤醒wait和sleep
 *  第五章 java中Future的使用
 *   *  创建Future
     *  从Future获取结果
     *  取消Future
     *  多线程环境中运行
 *  第六章 java并发中ExecutorService的使用
 *   *  创建ExecutorService
     *  为ExecutorService分配Tasks
     *  关闭ExecutorService
     *  Future
     *  ScheduledExecutorService
     *  ExecutorService和 Fork/Join
 *  第七章 java中Runnable和Callable的区别
 *   *  运行机制
     *  返回值的不同
     *  Exception处理
 *  第八章 ThreadLocal的使用
 *   *  在Map中存储用户数据
     *  在ThreadLocal中存储用户数据
 *  第九章 java中线程的生命周期
 *   *  java中Thread的状态
     *  NEW
     *  Runnable
     *  BLOCKED
     *  WAITING
     *  TIMED\_WAITING
     *  TERMINATED
 *  第十章 java中join的使用
 *  第十一章 怎么在java中关闭一个thread
 *  第十二章 java中的Atomic类
 *   *  问题背景
     *  Lock
     *  使用Atomic
 *  第十三章 java中interrupt，interrupted和isInterrupted的区别
 *   *  isInterrupted
     *  interrupted
     *  interrupt
 *  总结

并发是java高级程序员必须要深入研究的话题，从Synchronized到Lock，JDK本身提供了很多优秀的并发类和锁控制器，灵活使用这些类，可以写出优秀的并发程序，而这些类基本上都是在java.util.concurrent包中的，本文将会从具体的例子出发，一步一步带领大家进入java高质量并发的世界。

本文PDF下载链接[concurrent-all-in-one.pdf][]

本文的例子可以参考[https://github.com/ddean2009/learn-java-concurrency/][https_github.com_ddean2009_learn-java-concurrency]

# 第一章 java.util.concurrent简介 #

java.util.concurrent包提供了很多有用的类，方便我们进行并发程序的开发。本文将会做一个总体的简单介绍。

## 主要的组件 ##

java.util.concurrent包含了很多内容， 本文将会挑选其中常用的一些类来进行大概的说明：

*  Executor
 *  ExecutorService
 *  ScheduledExecutorService
 *  Future
 *  CountDownLatch
 *  CyclicBarrier
 *  Semaphore
 *  ThreadFactory

## Executor ##

Executor是一个接口，它定义了一个execute方法，这个方法接收一个Runnable，并在其中调用Runnable的run方法。

我们看一个Executor的实现：

public class Invoker implements Executor { 
        @Override
        public void execute(Runnable r) { 
            r.run();
        }
    }

现在我们可以直接调用该类中的方法：

public void execute() { 
            Executor executor = new Invoker();
            executor.execute( () -> { 
                log.info("{}", Thread.currentThread().toString());
            });
        }

> 注意，Executor并不一定要求执行的任务是异步的。

## ExecutorService ##

如果我们真正的需要使用多线程的话，那么就需要用到ExecutorService了。

ExecutorService管理了一个内存的队列，并定时提交可用的线程。

我们首先定义一个Runnable类：

public class Task implements Runnable { 
        @Override
        public void run() { 
            // task details
        }
    }

我们可以通过Executors来方便的创建ExecutorService：

ExecutorService executor = Executors.newFixedThreadPool(10);

上面创建了一个ThreadPool， 我们也可以创建单线程的ExecutorService：

ExecutorService executor =Executors.newSingleThreadExecutor();

我们这样提交task：

public void execute() {  
        executor.submit(new Task()); 
    }

因为ExecutorService维持了一个队列，所以它不会自动关闭， 我们需要调用executor.shutdown() 或者executor.shutdownNow()来关闭它。

如果想要判断ExecutorService中的线程在收到shutdown请求后是否全部执行完毕，可以调用如下的方法：

try { 
                executor.awaitTermination( 5l, TimeUnit.SECONDS );
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }

## ScheduledExecutorService ##

ScheduledExecutorService和ExecutorService很类似，但是它可以周期性的执行任务。

我们这样创建ScheduledExecutorService：

ScheduledExecutorService executorService
                    = Executors.newSingleThreadScheduledExecutor();

executorService的schedule方法，可以传入Runnable也可以传入Callable：

Future<String> future = executorService.schedule(() -> { 
 // ...
 return "Hello world";
 }, 1, TimeUnit.SECONDS);
 
 ScheduledFuture<?> scheduledFuture = executorService.schedule(() -> { 
 // ...
 }, 1, TimeUnit.SECONDS);

还有两个比较相近的方法：

scheduleAtFixedRate( Runnable command, long initialDelay, long period, TimeUnit unit )
    
    scheduleWithFixedDelay( Runnable command, long initialDelay, long delay, TimeUnit unit )

两者的区别是前者的period是以任务开始时间来计算的，后者是以任务结束时间来计算。

## Future ##

Future用来获取异步执行的结果。可以调用cancel(boolean mayInterruptIfRunning) 方法来取消线程的执行。

我们看下怎么得到一个Future对象：

public void invoke() { 
 ExecutorService executorService = Executors.newFixedThreadPool(10);
 
 Future<String> future = executorService.submit(() -> { 
 // ...
 Thread.sleep(10000l);
 return "Hello world";
 });
 }

我们看下怎么获取Future的结果：

if (future.isDone() && !future.isCancelled()) { 
        try { 
            str = future.get();
        } catch (InterruptedException | ExecutionException e) { 
            e.printStackTrace();
        }
    }

future还可以接受一个时间参数，超过指定的时间，将会报TimeoutException。

try { 
        future.get(10, TimeUnit.SECONDS);
    } catch (InterruptedException | ExecutionException | TimeoutException e) { 
        e.printStackTrace();
    }

## CountDownLatch ##

CountDownLatch是一个并发中很有用的类，CountDownLatch会初始化一个counter，通过这个counter变量，来控制资源的访问。我们会在后面的文章详细介绍。

## CyclicBarrier ##

CyclicBarrier和CountDownLatch很类似。CyclicBarrier主要用于多个线程互相等待的情况，可以通过调用await() 方法等待，知道达到要等的数量。

public class Task implements Runnable { 
     
        private CyclicBarrier barrier;
     
        public Task(CyclicBarrier barrier) { 
            this.barrier = barrier;
        }
     
        @Override
        public void run() { 
            try { 
                LOG.info(Thread.currentThread().getName() + 
                  " is waiting");
                barrier.await();
                LOG.info(Thread.currentThread().getName() + 
                  " is released");
            } catch (InterruptedException | BrokenBarrierException e) { 
                e.printStackTrace();
            }
        }
     
    }

public void start() { 
     
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> { 
            // ...
            LOG.info("All previous tasks are completed");
        });
     
        Thread t1 = new Thread(new Task(cyclicBarrier), "T1"); 
        Thread t2 = new Thread(new Task(cyclicBarrier), "T2"); 
        Thread t3 = new Thread(new Task(cyclicBarrier), "T3"); 
     
        if (!cyclicBarrier.isBroken()) {  
            t1.start(); 
            t2.start(); 
            t3.start(); 
        }
    }

## Semaphore ##

Semaphore包含了一定数量的许可证，通过获取许可证，从而获得对资源的访问权限。通过 tryAcquire()来获取许可，如果获取成功，许可证的数量将会减少。

一旦线程release()许可，许可的数量将会增加。

我们看下怎么使用：

static Semaphore semaphore = new Semaphore(10);
     
    public void execute() throws InterruptedException { 
     
        LOG.info("Available permit : " + semaphore.availablePermits());
        LOG.info("Number of threads waiting to acquire: " + 
          semaphore.getQueueLength());
     
        if (semaphore.tryAcquire()) { 
            try { 
                // ...
            }
            finally { 
                semaphore.release();
            }
        }
     
    }

## ThreadFactory ##

ThreadFactory可以很方便的用来创建线程：

public class ThreadFactoryUsage implements ThreadFactory { 
        private int threadId;
        private String name;
    
        public ThreadFactoryUsage(String name) { 
            threadId = 1;
            this.name = name;
        }
    
        @Override
        public Thread newThread(Runnable r) { 
            Thread t = new Thread(r, name + "-Thread_" + threadId);
            log.info("created new thread with id : " + threadId +
                    " and name : " + t.getName());
            threadId++;
            return t;
        }
    }

# 第二章 java并发中的Synchronized关键词 #

如果在多线程的环境中，我们经常会遇到资源竞争的情况，比如多个线程要去同时修改同一个共享变量，这时候，就需要对资源的访问方法进行一定的处理，保证同一时间只有一个线程访问。

java提供了synchronized关键字，方便我们实现上述操作。

## 为什么要同步 ##

我们举个例子，我们创建一个类，提供了一个setSum的方法：

public class SynchronizedMethods { 
    
        private int sum = 0;
    
        public void calculate() { 
            setSum(getSum() + 1);
        }
    }

如果我们在多线程的环境中调用这个calculate方法：

@Test
        public void givenMultiThread_whenNonSyncMethod() throws InterruptedException { 
            ExecutorService service = Executors.newFixedThreadPool(3);
            SynchronizedMethods summation = new SynchronizedMethods();
    
            IntStream.range(0, 1000)
                    .forEach(count -> service.submit(summation::calculate));
            service.shutdown();
            service.awaitTermination(1000, TimeUnit.MILLISECONDS);
    
            assertEquals(1000, summation.getSum());
        }

按照上面的方法，我们预计要返回1000， 但是实际上基本不可能得到1000这个值，因为在多线程环境中，对同一个资源进行同时操作带来的不利影响。

那我们怎么才能够建线程安全的环境呢？

## Synchronized关键词 ##

java提供了多种线程安全的方法，本文主要讲解Synchronized关键词，Synchronized关键词可以有很多种形式：

*  Instance methods
 *  Static methods
 *  Code blocks

当我们使用synchronized时，java会在相应的对象上加锁，从而在同一个对象等待锁的方法都必须顺序执行，从而保证了线程的安全。

### Synchronized Instance Methods ###

Synchronized关键词可以放在实例方法的前面：

public synchronized void synchronisedCalculate() { 
            setSum(getSum() + 1);
        }

看下调用结果：

@Test
    public void givenMultiThread_whenMethodSync() { 
        ExecutorService service = Executors.newFixedThreadPool(3);
        SynchronizedMethods method = new SynchronizedMethods();
     
        IntStream.range(0, 1000)
          .forEach(count -> service.submit(method::synchronisedCalculate));
        service.awaitTermination(1000, TimeUnit.MILLISECONDS);
     
        assertEquals(1000, method.getSum());
    }

这里synchronized将会锁住该方法的实例对象，多个线程中只有获得该实例对象锁的线程才能够执行。

### Synchronized Static Methods ###

Synchronized关键词也可以用在static方法前面：

public static synchronized void syncStaticCalculate() { 
            staticSum = staticSum + 1;
        }

Synchronized放在static方法前面和实例方法前面锁住的对象不同。放在static方法前面锁住的对象是这个Class本身，因为一个Class在JVM中只会存在一个，所以不管有多少该Class的实例，在同一时刻只会有一个线程可以执行该放方法。

@Test
        public void givenMultiThread_whenStaticSyncMethod() throws InterruptedException { 
            ExecutorService service = Executors.newCachedThreadPool();
    
            IntStream.range(0, 1000)
                    .forEach(count ->
                            service.submit(SynchronizedMethods::syncStaticCalculate));
            service.shutdown();
            service.awaitTermination(100, TimeUnit.MILLISECONDS);
    
            assertEquals(1000, SynchronizedMethods.staticSum);
        }

### Synchronized Blocks ###

有时候，我们可能不需要Synchronize整个方法，而是同步其中的一部分，这时候，我们可以使用Synchronized Blocks：

public void performSynchronizedTask() { 
            synchronized (this) { 
                setSum(getSum() + 1);
            }
        }

我们看下怎么测试：

@Test
        public void givenMultiThread_whenBlockSync() throws InterruptedException { 
            ExecutorService service = Executors.newFixedThreadPool(3);
            SynchronizedMethods synchronizedBlocks = new SynchronizedMethods();
    
            IntStream.range(0, 1000)
                    .forEach(count ->
                            service.submit(synchronizedBlocks::performSynchronizedTask));
            service.shutdown();
            service.awaitTermination(100, TimeUnit.MILLISECONDS);
    
            assertEquals(1000, synchronizedBlocks.getSum());
        }

上面我们同步的是实例，如果在静态方法中，我们也可以同步class：

public static void performStaticSyncTask(){ 
            synchronized (SynchronizedMethods.class) { 
                staticSum = staticSum + 1;
            }
        }

我们看下怎么测试：

@Test
        public void givenMultiThread_whenStaticSyncBlock() throws InterruptedException { 
            ExecutorService service = Executors.newCachedThreadPool();
    
            IntStream.range(0, 1000)
                    .forEach(count ->
                            service.submit(SynchronizedMethods::performStaticSyncTask));
            service.shutdown();
            service.awaitTermination(100, TimeUnit.MILLISECONDS);
    
            assertEquals(1000, SynchronizedMethods.staticSum);
        }

# 第三章 java中的Volatile关键字使用 #

在本文中，我们会介绍java中的一个关键字volatile。 volatile的中文意思是易挥发的，不稳定的。那么在java中使用是什么意思呢？

我们知道，在java中，每个线程都会有个自己的内存空间，我们称之为working memory。这个空间会缓存一些变量的信息，从而提升程序的性能。当执行完某个操作之后，thread会将更新后的变量更新到主缓存中，以供其他线程读写。

因为变量存在working memory和main memory两个地方，那么就有可能出现不一致的情况。 那么我们就可以使用Volatile关键字来强制将变量直接写到main memory，从而保证了不同线程读写到的是同一个变量。

## 什么时候使用volatile ##

那么我们什么时候使用volatile呢？当一个线程需要立刻读取到另外一个线程修改的变量值的时候，我们就可以使用volatile。我们来举个例子：

public class VolatileWithoutUsage { 
        private  int count = 0;
    
        public void incrementCount() { 
            count++;
        }
        public int getCount() { 
            return count;
        }
    }

这个类定义了一个incrementCount()方法，会去更新count值，我们接下来在多线程环境中去测试这个方法：

@Test
        public void testWithoutVolatile() throws InterruptedException { 
            ExecutorService service= Executors.newFixedThreadPool(3);
            VolatileWithoutUsage volatileWithoutUsage=new VolatileWithoutUsage();
    
            IntStream.range(0,1000).forEach(count ->service.submit(volatileWithoutUsage::incrementCount) );
            service.shutdown();
            service.awaitTermination(1000, TimeUnit.MILLISECONDS);
            assertEquals(1000,volatileWithoutUsage.getCount() );
        }

运行一下，我们会发现结果是不等于1000的。

java.lang.AssertionError: 
    Expected :1000
    Actual   :999

这是因为多线程去更新同一个变量，我们在上篇文章也提到了，这种情况可以通过加Synchronized关键字来解决。

那么是不是我们加上Volatile关键字后就可以解决这个问题了呢？

public class VolatileFalseUsage { 
        private volatile int count = 0;
    
        public void incrementCount() { 
            count++;
        }
        public int getCount() { 
            return count;
        }
    
    }

上面的类中，我们加上了关键字Volatile，我们再测试一下：

@Test
        public void testWithVolatileFalseUsage() throws InterruptedException { 
            ExecutorService service= Executors.newFixedThreadPool(3);
            VolatileFalseUsage volatileFalseUsage=new VolatileFalseUsage();
    
            IntStream.range(0,1000).forEach(count ->service.submit(volatileFalseUsage::incrementCount) );
            service.shutdown();
            service.awaitTermination(5000, TimeUnit.MILLISECONDS);
            assertEquals(1000,volatileFalseUsage.getCount() );
        }

运行一下，我们会发现结果还是错误的：

java.lang.AssertionError: 
    Expected :1000
    Actual   :992
    ~~
    
    为什么呢？ 我们先来看下count++的操作，count++可以分解为三步操作，1. 读取count的值，2.给count加1， 3.将count写回内存。添加Volatile关键词只能够保证count的变化立马可见，而不能保证1，2，3这三个步骤的总体原子性。 要实现总体的原子性还是需要用到类似Synchronized的关键字。
    
    下面看下正确的用法：
    
    ~~~java
    public class VolatileTrueUsage {
        private volatile int count = 0;
    
        public void setCount(int number) {
            count=number;
        }
        public int getCount() {
            return count;
        }
    }

@Test
        public void testWithVolatileTrueUsage() throws InterruptedException { 
            VolatileTrueUsage volatileTrueUsage=new VolatileTrueUsage();
            Thread threadA = new Thread(()->volatileTrueUsage.setCount(10));
            threadA.start();
            Thread.sleep(100);
    
            Thread reader = new Thread(() -> { 
                int valueReadByThread = volatileTrueUsage.getCount();
                assertEquals(10, valueReadByThread);
            });
            reader.start();
        }

## Happens-Before ##

从java5之后，volatile提供了一个Happens-Before的功能。Happens-Before 是指当volatile进行写回主内存的操作时，会将之前的非volatile的操作一并写回主内存。

public class VolatileHappenBeforeUsage { 
    
        int a = 0;
        volatile boolean flag = false;
    
        public void writer() { 
            a = 1;              // 1 线程A修改共享变量
            flag = true;        // 2 线程A写volatile变量
        }
    }

上面的例子中，a是一个非volatile变量，flag是一个volatile变量，但是由于happens-before的特性，a 将会表现的和volatile一样。

# 第四章 wait和sleep的区别 #

在本篇文章中，我们将会讨论一下java中wait()和sleep()方法的区别。并讨论一下怎么使用这两个方法。

## Wait和sleep的区别 ##

wait() 是Object中定义的native方法：

public final native void wait(long timeout) throws InterruptedException;

所以每一个类的实例都可以调用这个方法。wait()只能在synchronized block中调用。它会释放synchronized时加在object上的锁。

sleep()是定义Thread中的native静态类方法：

public static native void sleep(long millis) throws InterruptedException;

所以Thread.sleep()可以在任何情况下调用。Thread.sleep()将会暂停当前线程，并且不会释放任何锁资源。

我们先看一下一个简单的wait使用：

@Slf4j
    public class WaitUsage { 
    
        private static Object LOCK = new Object();
    
        public static void WaitExample() throws InterruptedException { 
            synchronized (LOCK) { 
                LOCK.wait(1000);
                log.info("Object '" + LOCK + "' is woken after" +
                        " waiting for 1 second");
            }
        }
    }

再看一下sleep的使用：

@Slf4j
    public class SleepUsage { 
    
        public static void sleepExample() throws InterruptedException { 
            Thread.sleep(1000);
            log.info(
                    "Thread '" + Thread.currentThread().getName() +
                            "' is woken after sleeping for 1 second");
        }
    }

## 唤醒wait和sleep ##

sleep()方法自带sleep时间，时间过后，Thread会自动被唤醒。  
或者可以通过调用interrupt()方法来中断。

相比而言wait的唤醒会比较复杂，我们需要调用notify() 和 notifyAll()方法来唤醒等待在特定wait object上的线程。

notify()会根据线程调度的机制选择一个线程来唤醒，而notifyAll()会唤醒所有等待的线程，由这些线程重新争夺资源锁。

wait,notity通常用在生产者和消费者情形，我们看下怎么使用：

@Slf4j
    public class WaitNotifyUsage { 
    
        private int count =0;
    
        public void produceMessage() throws InterruptedException { 
    
            while(true) { 
                synchronized (this) { 
                    while (count == 5) { 
                        log.info("count == 5 , wait ....");
                        wait();
                    }
                    count++;
                    log.info("produce count {}", count);
                    notify();
                }
            }
        }
    
        public void consumeMessage() throws InterruptedException { 
    
            while (true) { 
                synchronized (this) { 
                    while (count == 0) { 
                        log.info("count == 0, wait ...");
                        wait();
                    }
                    log.info("consume count {}", count);
                    count--;
                    notify();
                }
            }
        }
    }

看下怎么调用：

@Test
        public void testWaitNotifyUsage() throws InterruptedException{ 
            WaitNotifyUsage waitNotifyUsage=new WaitNotifyUsage();
    
            ExecutorService executorService=Executors.newFixedThreadPool(4);
            executorService.submit(()-> { 
                try { 
                    waitNotifyUsage.produceMessage();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
    
            executorService.submit(()-> { 
                try { 
                    waitNotifyUsage.consumeMessage();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
    
            Thread.sleep(50000);
        }

# 第五章 java中Future的使用 #

Future是java 1.5引入的一个interface，可以方便的用于异步结果的获取。 本文将会通过具体的例子讲解如何使用Future。

## 创建Future ##

正如上面所说，Future代表的是异步执行的结果，意思是当异步执行结束之后，返回的结果将会保存在Future中。

那么我们什么时候会用到Future呢？ 一般来说，当我们执行一个长时间运行的任务时，使用Future就可以让我们暂时去处理其他的任务，等长任务执行完毕再返回其结果。

经常会使用到Future的场景有：1. 计算密集场景。2. 处理大数据量。3. 远程方法调用等。

接下来我们将会使用ExecutorService来创建一个Future。

<T> Future<T> submit(Callable<T> task);

上面是ExecutorService中定义的一个submit方法，它接收一个Callable参数，并返回一个Future。

我们用一个线程来计算一个平方运算：

private ExecutorService executor
 = Executors.newSingleThreadExecutor();
 
 public Future<Integer> calculate(Integer input) { 
 return executor.submit(() -> { 
 System.out.println("Calculating..."+ input);
 Thread.sleep(1000);
 return input * input;
 });
 }

submit需要接受一个Callable参数，Callable需要实现一个call方法，并返回结果。这里我们使用lamaba表达式来简化这一个流程。

## 从Future获取结果 ##

上面我们创建好了Future，接下来我们看一下怎么获取到Future的值。

FutureUsage futureUsage=new FutureUsage();
 Future<Integer> futureOne = futureUsage.calculate(20);
 while(!futureOne.isDone()) { 
 System.out.println("Calculating...");
 Thread.sleep(300);
 }
 Integer result = futureOne.get();

首先我们通过Future.isDone() 来判断这个异步操作是否执行完毕，如果完毕我们就可以直接调用futureOne.get()来获得Futre的结果。

这里futureOne.get()是一个阻塞操作，会一直等待异步执行完毕才返回结果。

如果我们不想等待，future提供了一个带时间的方法：

Integer result = futureOne.get(500, TimeUnit.MILLISECONDS);

如果在等待时间结束的时候，Future还有返回，则会抛出一个TimeoutException。

## 取消Future ##

如果我们提交了一个异步程序，但是想取消它， 则可以这样：

uture<Integer> futureTwo = futureUsage.calculate(4);
 
 boolean canceled = futureTwo.cancel(true);

Future.cancel(boolean) 传入一个boolean参数，来选择是否中断正在运行的task。

如果我们cancel之后，再次调用get()方法，则会抛出CancellationException。

## 多线程环境中运行 ##

如果有两个计算任务，先看下在单线程下运行的结果。

Future<Integer> future1 = futureUsage.calculate(10);
 Future<Integer> future2 = futureUsage.calculate(100);
 
 while (!(future1.isDone() && future2.isDone())) { 
 System.out.println(
 String.format(
 "future1 is %s and future2 is %s",
 future1.isDone() ? "done" : "not done",
 future2.isDone() ? "done" : "not done"
 )
 );
 Thread.sleep(300);
 }
 
 Integer result1 = future1.get();
 Integer result2 = future2.get();
 
 System.out.println(result1 + " and " + result2);

因为我们通过Executors.newSingleThreadExecutor（）来创建的单线程池。所以运行结果如下：

Calculating...10
    future1 is not done and future2 is not done
    future1 is not done and future2 is not done
    future1 is not done and future2 is not done
    future1 is not done and future2 is not done
    Calculating...100
    future1 is done and future2 is not done
    future1 is done and future2 is not done
    future1 is done and future2 is not done
    100 and 10000

如果我们使用Executors.newFixedThreadPool(2)来创建一个多线程池，则可以得到如下的结果：

calculating...10
    calculating...100
    future1 is not done and future2 is not done
    future1 is not done and future2 is not done
    future1 is not done and future2 is not done
    future1 is not done and future2 is not done
    100 and 10000

# 第六章 java并发中ExecutorService的使用 #

ExecutorService是java中的一个异步执行的框架，通过使用ExecutorService可以方便的创建多线程执行环境。

本文将会详细的讲解ExecutorService的具体使用。

## 创建ExecutorService ##

通常来说有两种方法来创建ExecutorService。

第一种方式是使用Executors中的工厂类方法，例如：

ExecutorService executor = Executors.newFixedThreadPool(10);

除了newFixedThreadPool方法之外，Executors还包含了很多创建ExecutorService的方法。

第二种方法是直接创建一个ExecutorService， 因为ExecutorService是一个interface，我们需要实例化ExecutorService的一个实现。

这里我们使用ThreadPoolExecutor来举例：

ExecutorService executorService =
 new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>());

## 为ExecutorService分配Tasks ##

ExecutorService可以执行Runnable和Callable的task。其中Runnable是没有返回值的，而Callable是有返回值的。我们分别看一下两种情况的使用：

Runnable runnableTask = () -> { 
 try { 
 TimeUnit.MILLISECONDS.sleep(300);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 }
 };
 
 Callable<String> callableTask = () -> { 
 TimeUnit.MILLISECONDS.sleep(300);
 return "Task's execution";
 };

将task分配给ExecutorService，可以通过调用xecute(), submit(), invokeAny(), invokeAll()这几个方法来实现。

execute() 返回值是void，他用来提交一个Runnable task。

executorService.execute(runnableTask);

submit() 返回值是Future，它可以提交Runnable task, 也可以提交Callable task。 提交Runnable的有两个方法：

<T> Future<T> submit(Runnable task, T result);
 
 Future<?> submit(Runnable task);

第一个方法在返回传入的result。第二个方法返回null。

再看一下callable的使用：

Future<String> future = 
 executorService.submit(callableTask);

invokeAny() 将一个task列表传递给executorService，并返回其中的一个成功返回的结果。

String result = executorService.invokeAny(callableTasks);

invokeAll() 将一个task列表传递给executorService，并返回所有成功执行的结果：

List<Future<String>> futures = executorService.invokeAll(callableTasks);

## 关闭ExecutorService ##

如果ExecutorService中的任务运行完毕之后，ExecutorService不会自动关闭。它会等待接收新的任务。如果需要关闭ExecutorService， 我们需要调用shutdown() 或者 shutdownNow() 方法。

shutdown() 会立即销毁ExecutorService，它会让ExecutorServic停止接收新的任务，并等待现有任务全部执行完毕再销毁。

executorService.shutdown();

shutdownNow()并不保证所有的任务都被执行完毕，它会返回一个未执行任务的列表：

List<Runnable> notExecutedTasks = executorService.shutdownNow();

oracle推荐的最佳关闭方法是和awaitTermination一起使用：

executorService.shutdown();
           try { 
               if (!executorService.awaitTermination(800, TimeUnit.MILLISECONDS)) { 
                   executorService.shutdownNow();
               }
           } catch (InterruptedException e) { 
               executorService.shutdownNow();
           }

先停止接收任务，然后再等待一定的时间让所有的任务都执行完毕，如果超过了给定的时间，则立刻结束任务。

## Future ##

submit() 和 invokeAll() 都会返回Future对象。之前的文章我们已经详细讲过了Future。 这里就只列举一下怎么使用：

Future<String> future = executorService.submit(callableTask);
 String result = null;
 try { 
 result = future.get();
 } catch (InterruptedException | ExecutionException e) { 
 e.printStackTrace();
 }

## ScheduledExecutorService ##

ScheduledExecutorService为我们提供了定时执行任务的机制。

我们这样创建ScheduledExecutorService：

ScheduledExecutorService executorService
                    = Executors.newSingleThreadScheduledExecutor();

executorService的schedule方法，可以传入Runnable也可以传入Callable：

还有两个比较相近的方法：

scheduleAtFixedRate( Runnable command, long initialDelay, long period, TimeUnit unit )
    
    scheduleWithFixedDelay( Runnable command, long initialDelay, long delay, TimeUnit unit )

两者的区别是前者的period是以任务开始时间来计算的，后者是以任务结束时间来计算。

## ExecutorService和 Fork/Join ##

java 7 引入了Fork/Join框架。 那么两者的区别是什么呢？

ExecutorService可以由用户来自己控制生成的线程，提供了对线程更加细粒度的控制。而Fork/Join则是为了让任务更加快速的执行完毕。

# 第七章 java中Runnable和Callable的区别 #

在java的多线程开发中Runnable一直以来都是多线程的核心，而Callable是java1.5添加进来的一个增强版本。

本文我们会详细探讨Runnable和Callable的区别。

## 运行机制 ##

首先看下Runnable和Callable的接口定义：

@FunctionalInterface
 public interface Runnable { 
 /** * When an object implementing interface <code>Runnable</code> is used * to create a thread, starting the thread causes the object's * <code>run</code> method to be called in that separately executing * thread. * * The general contract of the method <code>run</code> is that it may * take any action whatsoever. * * @see java.lang.Thread#run() */
 public abstract void run();
 }

@FunctionalInterface
 public interface Callable<V> { 
 /** * Computes a result, or throws an exception if unable to do so. * * @return computed result * @throws Exception if unable to compute a result */
 V call() throws Exception;
 }

Runnable需要实现run（）方法，Callable需要实现call（）方法。

我们都知道要自定义一个Thread有两种方法，一是继承Thread，而是实现Runnable接口，这是因为Thread本身就是一个Runnable的实现：

class Thread implements Runnable { 
 /* Make sure registerNatives is the first thing <clinit> does. */
 private static native void registerNatives();
 static { 
 registerNatives();
 }
 ...

所以Runnable可以通过Runnable和之前我们介绍的ExecutorService 来执行，而Callable则只能通过ExecutorService 来执行。

## 返回值的不同 ##

根据上面两个接口的定义，Runnable是不返还值的，而Callable可以返回值。

如果我们都通过ExecutorService来提交，看看有什么不同：

*  使用runnable

public void executeTask() { 
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            Future future = executorService.submit(()->log.info("in runnable!!!!"));
            executorService.shutdown();
        }

*  使用callable

public void executeTask() { 
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            Future future = executorService.submit(()->{ 
                log.info("in callable!!!!");
                return "callable";
            });
            executorService.shutdown();
        }

虽然我们都返回了Future，但是runnable的情况下Future将不包含任何值。

## Exception处理 ##

Runnable的run（）方法定义没有抛出任何异常，所以任何的Checked Exception都需要在run（）实现方法中自行处理。

Callable的Call（）方法抛出了throws Exception，所以可以在call（）方法的外部，捕捉到Checked Exception。我们看下Callable中异常的处理。

public void executeTaskWithException(){ 
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            Future future = executorService.submit(()->{ 
                log.info("in callable!!!!");
                throw new CustomerException("a customer Exception");
            });
            try { 
                Object object= future.get();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
                e.getCause();
            }
            executorService.shutdown();
        }

上面的例子中，我们在Callable中抛出了一个自定义的CustomerException。

这个异常会被包含在返回的Future中。当我们调用future.get()方法时，就会抛出ExecutionException，通过e.getCause()，就可以获取到包含在里面的具体异常信息。

# 第八章 ThreadLocal的使用 #

ThreadLocal主要用来为当前线程存储数据，这个数据只有当前线程可以访问。

在定义ThreadLocal的时候，我们可以同时定义存储在ThreadLocal中的特定类型的对象。

ThreadLocal<Integer> threadLocalValue = new ThreadLocal<>();

上面我们定义了一个存储Integer的ThreadLocal对象。

要存储和获取ThreadLocal中的对象也非常简单，使用get（）和set（）即可：

threadLocalValue.set(1);
    Integer result = threadLocalValue.get();

我可以将ThreadLocal看成是一个map，而当前的线程就是map中的key。

除了new一个ThreadLocal对象，我们还可以通过：

public static <S> ThreadLocal<S> withInitial(Supplier<? extends S> supplier) { 
 return new SuppliedThreadLocal<>(supplier);
 }

ThreadLocal提供的静态方法withInitial来初始化一个ThreadLocal。

ThreadLocal<Integer> threadLocal = ThreadLocal.withInitial(() -> 1);

withInitial需要一个Supplier对象，通过调用Supplier的get()方法获取到初始值。

要想删除ThreadLocal中的存储数据，可以调用：

threadLocal.remove();

下面我通过两个例子的对比，来看一下使用ThreadLocal的好处。

在实际的应用中，我们通常会需要为不同的用户请求存储不同的用户信息，一般来说我们需要构建一个全局的Map，来根据不同的用户ID，来存储不同的用户信息，方便在后面获取。

## 在Map中存储用户数据 ##

我们先看下如果使用全局的Map该怎么用：

public class SharedMapWithUserContext implements Runnable { 
 
 public static Map<Integer, Context> userContextPerUserId
 = new ConcurrentHashMap<>();
 private Integer userId;
 private UserRepository userRepository = new UserRepository();
 
 public SharedMapWithUserContext(int i) { 
 this.userId=i;
 }
 
 @Override
 public void run() { 
 String userName = userRepository.getUserNameForUserId(userId);
 userContextPerUserId.put(userId, new Context(userName));
 }
 }

这里我们定义了一个static的Map来存取用户信息。

再看一下怎么使用：

@Test
        public void testWithMap(){ 
            SharedMapWithUserContext firstUser = new SharedMapWithUserContext(1);
            SharedMapWithUserContext secondUser = new SharedMapWithUserContext(2);
            new Thread(firstUser).start();
            new Thread(secondUser).start();
            assertEquals(SharedMapWithUserContext.userContextPerUserId.size(), 2);
        }

## 在ThreadLocal中存储用户数据 ##

如果我们要在ThreadLocal中使用可以这样：

public class ThreadLocalWithUserContext implements Runnable { 
 
 private static ThreadLocal<Context> userContext
 = new ThreadLocal<>();
 private Integer userId;
 private UserRepository userRepository = new UserRepository();
 
 public ThreadLocalWithUserContext(int i) { 
 this.userId=i;
 }
 
 @Override
 public void run() { 
 String userName = userRepository.getUserNameForUserId(userId);
 userContext.set(new Context(userName));
 System.out.println("thread context for given userId: "
 + userId + " is: " + userContext.get());
 }
 
 }

测试代码如下：

public class ThreadLocalWithUserContextTest { 
    
        @Test
        public void testWithThreadLocal(){ 
            ThreadLocalWithUserContext firstUser
                    = new ThreadLocalWithUserContext(1);
            ThreadLocalWithUserContext secondUser
                    = new ThreadLocalWithUserContext(2);
            new Thread(firstUser).start();
            new Thread(secondUser).start();
        }
    }

运行之后，我们可以得到下面的结果：

thread context for given userId: 1 is: com.flydean.Context@411734d4
    thread context for given userId: 2 is: com.flydean.Context@1e9b6cc

不同的用户信息被存储在不同的线程环境中。

> 注意，我们使用ThreadLocal的时候，一定是我们可以自由的控制所创建的线程。如果在ExecutorService环境下，就最好不要使用ThreadLocal，因为在ExecutorService中，线程是不可控的。

# 第九章 java中线程的生命周期 #

线程是java中绕不过去的一个话题， 今天本文将会详细讲解java中线程的生命周期，希望可以给大家一些启发。

## java中Thread的状态 ##

java中Thread有6种状态，分别是：

1.  NEW - 新创建的Thread，还没有开始执行
2.  RUNNABLE - 可运行状态的Thread，包括准备运行和正在运行的。
3.  BLOCKED - 正在等待资源锁的线程
4.  WAITING - 正在无限期等待其他线程来执行某个特定操作
5.  TIMED\_WAITING - 在一定的时间内等待其他线程来执行某个特定操作
6.  TERMINATED - 线程执行完毕

我们可以用一个图来直观的表示：

![eb7ae88ab10b06c14e3017c2799038bf.png][]

JDK代码中的定义如下：

public enum State { 
 /** * Thread state for a thread which has not yet started. */
 NEW,
 
 /** * Thread state for a runnable thread. A thread in the runnable * state is executing in the Java virtual machine but it may * be waiting for other resources from the operating system * such as processor. */
 RUNNABLE,
 
 /** * Thread state for a thread blocked waiting for a monitor lock. * A thread in the blocked state is waiting for a monitor lock * to enter a synchronized block/method or * reenter a synchronized block/method after calling * {@link Object#wait() Object.wait}. */
 BLOCKED,
 
 /** * Thread state for a waiting thread. * A thread is in the waiting state due to calling one of the * following methods: * <ul> * <li>{@link Object#wait() Object.wait} with no timeout</li> * <li>{@link #join() Thread.join} with no timeout</li> * <li>{@link LockSupport#park() LockSupport.park}</li> * </ul> * * A thread in the waiting state is waiting for another thread to * perform a particular action. * * For example, a thread that has called <tt>Object.wait()</tt> * on an object is waiting for another thread to call * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on * that object. A thread that has called <tt>Thread.join()</tt> * is waiting for a specified thread to terminate. */
 WAITING,
 
 /** * Thread state for a waiting thread with a specified waiting time. * A thread is in the timed waiting state due to calling one of * the following methods with a specified positive waiting time: * <ul> * <li>{@link #sleep Thread.sleep}</li> * <li>{@link Object#wait(long) Object.wait} with timeout</li> * <li>{@link #join(long) Thread.join} with timeout</li> * <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li> * <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li> * </ul> */
 TIMED_WAITING,
 
 /** * Thread state for a terminated thread. * The thread has completed execution. */
 TERMINATED;
 }

## NEW ##

NEW 表示线程创建了，但是还没有开始执行。我们看一个NEW的例子：

public class NewThread implements Runnable{ 
        public static void main(String[] args) { 
            Runnable runnable = new NewThread();
            Thread t = new Thread(runnable);
            log.info(t.getState().toString());
        }
    
        @Override
        public void run() { 
    
        }
    }

上面的代码将会输出：

NEW

## Runnable ##

Runnable表示线程正在可执行状态。包括正在运行和准备运行两种。

为什么这两种都叫做Runnable呢？我们知道在多任务环境中，CPU的个数是有限的，所以任务都是轮循占有CPU来处理的，JVM中的线程调度器会为每个线程分配特定的执行时间，当执行时间结束后，线程调度器将会释放CPU，以供其他的Runnable线程执行。

我们看一个Runnable的例子：

public class RunnableThread implements Runnable { 
        @Override
        public void run() { 
    
        }
    
        public static void main(String[] args) { 
            Runnable runnable = new RunnableThread();
            Thread t = new Thread(runnable);
            t.start();
            log.info(t.getState().toString());
        }
    }

上面的代码将会输出：

RUNNABLE

## BLOCKED ##

BLOCKED表示线程正在等待资源锁，而目前该资源正在被其他线程占有。

我们举个例子：

public class BlockThread implements Runnable { 
        @Override
        public void run() { 
            loopResource();
        }
    
        public static synchronized void loopResource() { 
            while(true) { 
                //无限循环
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            Thread t1 = new Thread(new BlockThread());
            Thread t2 = new Thread(new BlockThread());
    
            t1.start();
            t2.start();
    
            Thread.sleep(1000);
            log.info(t1.getState().toString());
            log.info(t2.getState().toString());
            System.exit(0);
        }
    }

上面的例子中，由于t1是无限循环，将会一直占有资源锁，导致t2无法获取资源锁，从而位于BLOCKED状态。

我们会得到如下结果：

12:40:11.710 [main] INFO com.flydean.BlockThread - RUNNABLE
    12:40:11.713 [main] INFO com.flydean.BlockThread - BLOCKED

## WAITING ##

WAITING 状态表示线程正在等待其他的线程执行特定的操作。有三种方法可以导致线程处于WAITTING状态：

1.  object.wait()
2.  thread.join()
3.  LockSupport.park()

其中1，2方法不需要传入时间参数。

我们看下使用的例子：

public class WaitThread implements  Runnable{ 
    
        public static Thread t1;
        @Override
        public void run() { 
            Thread t2 = new Thread(()->{ 
                try { 
                    Thread.sleep(10000);
                } catch (InterruptedException e) { 
                    Thread.currentThread().interrupt();
                    log.error("Thread interrupted", e);
                }
                log.info("t1"+t1.getState().toString());
            });
            t2.start();
    
            try { 
                t2.join();
            } catch (InterruptedException e) { 
                Thread.currentThread().interrupt();
                log.error("Thread interrupted", e);
            }
            log.info("t2"+t2.getState().toString());
        }
    
        public static void main(String[] args) { 
            t1 = new Thread(new WaitThread());
            t1.start();
    
        }
    }

在这个例子中，我们调用的t2.join()，这会使调用它的t1线程处于WAITTING状态。

我们看下输出结果：

12:44:12.958 [Thread-1] INFO com.flydean.WaitThread - t1 WAITING
    12:44:12.964 [Thread-0] INFO com.flydean.WaitThread - t2 TERMINATED

## TIMED\_WAITING ##

TIMED\_WAITING状态表示在一个有限的时间内等待其他线程执行特定的某些操作。

java中有5中方式来达到这种状态：

1.  thread.sleep(long millis)
2.  wait(int timeout) 或者 wait(int timeout, int nanos)
3.  thread.join(long millis)
4.  LockSupport.parkNanos
5.  LockSupport.parkUntil

我们举个例子：

public class TimedWaitThread implements  Runnable{ 
        @Override
        public void run() { 
            try { 
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                Thread.currentThread().interrupt();
                log.error("Thread interrupted", e);
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            TimedWaitThread obj1 = new TimedWaitThread();
            Thread t1 = new Thread(obj1);
            t1.start();
    
            // The following sleep will give enough time for ThreadScheduler
            // to start processing of thread t1
            Thread.sleep(1000);
            log.info(t1.getState().toString());
        }
    }

上面的例子中我们调用了Thread.sleep(5000)来让线程处于TIMED\_WAITING状态。

看下输出：

12:58:02.706 [main] INFO com.flydean.TimedWaitThread - TIMED_WAITING

那么问题来了，TIMED\_WAITING和WAITTING有什么区别呢？

TIMED\_WAITING如果在给定的时间内没有等到其他线程的特定操作，则会被唤醒，从而进入争夺资源锁的队列，如果能够获取到锁，则会变成Runnable状态，如果获取不到锁，则会变成BLOCKED状态。

## TERMINATED ##

TERMINATED表示线程已经执行完毕。我们看下例子：

public class TerminatedThread implements Runnable{ 
        @Override
        public void run() { 
    
        }
    
        public static void main(String[] args) throws InterruptedException { 
            Thread t1 = new Thread(new TerminatedThread());
            t1.start();
            // The following sleep method will give enough time for
            // thread t1 to complete
            Thread.sleep(1000);
            log.info(t1.getState().toString());
        }
    }

输出结果：

13:02:38.868 [main] INFO com.flydean.TerminatedThread - TERMINATED

# 第十章 java中join的使用 #

join()应该是我们在java中经常会用到的一个方法，它主要是将当前线程置为WAITTING状态，然后等待调用的线程执行完毕或被interrupted。

join()是Thread中定义的方法，我们看下他的定义：

/** * Waits for this thread to die. * * An invocation of this method behaves in exactly the same * way as the invocation * * <blockquote> * {@linkplain #join(long) join}{@code (0)} * </blockquote> * * @throws InterruptedException * if any thread has interrupted the current thread. The * interrupted status of the current thread is * cleared when this exception is thrown. */
 public final void join() throws InterruptedException { 
 join(0);
 }

我们看下join是怎么使用的，通常我们需要在线程A中调用线程B.join():

public class JoinThread implements Runnable{ 
        public int processingCount = 0;
    
        JoinThread(int processingCount) { 
            this.processingCount = processingCount;
            log.info("Thread Created");
        }
    
        @Override
        public void run() { 
            log.info("Thread " + Thread.currentThread().getName() + " started");
            while (processingCount > 0) { 
                try { 
                    Thread.sleep(1000);
                } catch (InterruptedException e) { 
                    log.info("Thread " + Thread.currentThread().getName() + " interrupted");
                }
                processingCount--;
            }
            log.info("Thread " + Thread.currentThread().getName() + " exiting");
        }
    
        @Test
        public void joinTest()
                throws InterruptedException { 
            Thread t2 = new Thread(new JoinThread(1));
            t2.start();
            log.info("Invoking join");
            t2.join();
            log.info("Returned from join");
            log.info("t2 status {}",t2.isAlive());
        }
    }

我们在主线程中调用了t2.join(),则主线程将会等待t2执行完毕，我们看下输出结果：

06:17:14.775 [main] INFO com.flydean.JoinThread - Thread Created
    06:17:14.779 [main] INFO com.flydean.JoinThread - Invoking join
    06:17:14.779 [Thread-0] INFO com.flydean.JoinThread - Thread Thread-0 started
    06:17:15.783 [Thread-0] INFO com.flydean.JoinThread - Thread Thread-0 exiting
    06:17:15.783 [main] INFO com.flydean.JoinThread - Returned from join
    06:17:15.783 [main] INFO com.flydean.JoinThread - t2 status false

当线程已经执行完毕或者还没开始执行的时候，join（）将会立即返回：

Thread t1 = new SampleThread(0);
    t1.join();  //returns immediately

join还有两个带时间参数的方法：

public final void join(long millis) throws InterruptedException

public final void join(long millis,int nanos) throws InterruptedException

如果在给定的时间内调用的线程没有返回，则主线程将会继续执行：

@Test
        public void testJoinTimeout()
                throws InterruptedException { 
            Thread t3 =  new Thread(new JoinThread(10));
            t3.start();
            t3.join(1000);
            log.info("t3 status {}", t3.isAlive());
        }

上面的例子将会输出：

06:30:58.159 [main] INFO com.flydean.JoinThread - Thread Created
    06:30:58.163 [Thread-0] INFO com.flydean.JoinThread - Thread Thread-0 started
    06:30:59.172 [main] INFO com.flydean.JoinThread - t3 status true

Join()还有个happen-before的特性，这就是如果thread t1调用 t2.join(), 那么当t2返回时，所有t2的变动都会t1可见。

之前我们讲volatile关键词的时候也提到了这个happen-before规则。 我们看下例子：

@Test
        public void testHappenBefore() throws InterruptedException { 
            JoinThread t4 =  new JoinThread(10);
            t4.start();
            // not guaranteed to stop even if t4 finishes.
            do { 
                log.info("inside the loop");
                Thread.sleep(1000);
            } while ( t4.processingCount > 0);
        }

我们运行下，可以看到while循环一直在进行中，即使t4中的变量已经变成了0。

所以如果我们需要在这种情况下使用的话，我们需要用到join（），或者其他的同步机制。

# 第十一章 怎么在java中关闭一个thread #

我们经常需要在java中用到thread，我们知道thread有一个start()方法可以开启一个线程。那么怎么关闭这个线程呢？

有人会说可以用Thread.stop（）方法。但是这个方法已经被废弃了。

根据Oracle的官方文档，Thread.stop是不安全的。因为调用stop方法的时候，将会释放它获取的所有监视器锁（通过传递ThreadDeath异常实现）。如果有资源该监视器锁所保护的话，就可能会出现数据不一致的异常。并且这种异常很难被发现。 所以现在已经不推荐是用Thread.stop方法了。

那我们还有两种方式来关闭一个Thread。

1.  Flag变量

如果我们有一个无法自动停止的Thread，我们可以创建一个条件变量，通过不断判断该变量的值，来决定是否结束该线程的运行。

public class KillThread implements Runnable { 
        private Thread worker;
        private final AtomicBoolean running = new AtomicBoolean(false);
        private int interval;
    
        public KillThread(int sleepInterval) { 
            interval = sleepInterval;
        }
    
        public void start() { 
            worker = new Thread(this);
            worker.start();
        }
    
        public void stop() { 
            running.set(false);
        }
    
        public void run() { 
            running.set(true);
            while (running.get()) { 
                try { 
                    Thread.sleep(interval);
                } catch (InterruptedException e){ 
                    Thread.currentThread().interrupt();
                    log.info("Thread was interrupted, Failed to complete operation");
                }
                // do something here
            }
            log.info("finished");
        }
    
        public static void main(String[] args) { 
            KillThread killThread= new KillThread(1000);
            killThread.start();
            killThread.stop();
        }
    
    
    }

上面的例子中，我们通过定义一个AtomicBoolean 的原子变量来存储Flag标志。

我们将会在后面的文章中详细的讲解原子变量。

1.  调用interrupt()方法

通过调用interrupt()方法，将会中断正在等待的线程，并抛出InterruptedException异常。

根据Oracle的说明，如果你想自己处理这个异常的话，需要reasserts出去，注意，这里是reasserts而不是rethrows，因为有些情况下，无法rethrow这个异常，我们需要这样做：

Thread.currentThread().interrupt();

这将会reasserts InterruptedException异常。

看下我们第二种方法怎么调用：

public class KillThread implements Runnable { 
        private Thread worker;
        private final AtomicBoolean running = new AtomicBoolean(false);
        private int interval;
    
        public KillThread(int sleepInterval) { 
            interval = sleepInterval;
        }
    
        public void start() { 
            worker = new Thread(this);
            worker.start();
        }
    
        public void interrupt() { 
            running.set(false);
            worker.interrupt();
        }
    
        public void stop() { 
            running.set(false);
        }
    
        public void run() { 
            running.set(true);
            while (running.get()) { 
                try { 
                    Thread.sleep(interval);
                } catch (InterruptedException e){ 
                    Thread.currentThread().interrupt();
                    log.info("Thread was interrupted, Failed to complete operation");
                }
                // do something here
            }
            log.info("finished");
        }
    
        public static void main(String[] args) { 
            KillThread killThread= new KillThread(1000);
            killThread.start();
            killThread.interrupt();
        }
    }

上面的例子中，当线程在Sleep中时，调用了interrupt方法，sleep会退出，并且抛出InterruptedException异常。

# 第十二章 java中的Atomic类 #

## 问题背景 ##

在多线程环境中，我们最常遇到的问题就是变量的值进行同步。因为变量需要在多线程中进行共享，所以我们必须需要采用一定的同步机制来进行控制。

通过之前的文章，我们知道可以采用Lock的机制，当然也包括今天我们讲的Atomic类。

下面我们从两种方式来分别介绍。

## Lock ##

在之前的文章中，我们也讲了同步的问题，我们再回顾一下。 如果定义了一个计数器如下：

public class Counter { 
    
        int counter;
    
        public void increment() { 
            counter++;
        }
    
    }

如果是在单线程环境中，上面的代码没有任何问题。但是如果在多线程环境中，counter++将会得到不同的结果。

因为虽然counter++看起来是一个原子操作，但是它实际上包含了三个操作：读数据，加一，写回数据。

我们之前的文章也讲了，如何解决这个问题：

public class LockCounter { 
    
        private volatile int counter;
    
        public synchronized void increment() { 
            counter++;
        }
    }

通过加synchronized，保证同一时间只会有一个线程去读写counter变量。

通过volatile，保证所有的数据直接操作的主缓存，而不使用线程缓存。

这样虽然解决了问题，但是性能可能会受影响，因为synchronized会锁住整个LockCounter实例。

## 使用Atomic ##

通过引入低级别的原子化语义命令（比如compare-and-swap (CAS)），从而能在保证效率的同时保证原子性。

一个标准的CAS包含三个操作：

1.  将要操作的内存地址M。
2.  现有的变量A。
3.  新的需要存储的变量B。

CAS将会先比较A和M中存储的值是否一致，一致则表示其他线程未对该变量进行修改，则将其替换为B。 否则不做任何操作。

使用CAS可以不用阻塞其他的线程，但是我们需要自己处理好当更新失败的情况下的业务逻辑处理情况。

Java提供了很多Atomic类，最常用的包括AtomicInteger, AtomicLong, AtomicBoolean, 和 AtomicReference.

其中的主要方法：

1.  get() – 直接中主内存中读取变量的值，类似于volatile变量。
2.  set() – 将变量写回主内存。类似于volatile变量。
3.  lazySet() – 延迟写回主内存。一种常用的情景是将引用重置为null的情况。
4.  compareAndSet() – 执行CAS操作，成功返回true，失败返回false。
5.  weakCompareAndSet() – 比较弱的CAS操作，不同的是它不执行happens-before操作，从而不保证能够读取到其他变量最新的值。

我们看下怎么用：

public class AtomicCounter { 
    
        private final AtomicInteger counter = new AtomicInteger(0);
    
        public int getValue() { 
            return counter.get();
        }
        public void increment() { 
            while(true) { 
                int existingValue = getValue();
                int newValue = existingValue + 1;
                if(counter.compareAndSet(existingValue, newValue)) { 
                    return;
                }
            }
        }
    }

# 第十三章 java中interrupt，interrupted和isInterrupted的区别 #

前面的文章我们讲到了调用interrupt()来停止一个Thread，本文将会详细讲解java中三个非常相似的方法interrupt，interrupted和isInterrupted。

## isInterrupted ##

首先看下最简单的isInterrupted方法。isInterrupted是Thread类中的一个实例方法：

public boolean isInterrupted() { 
            return isInterrupted(false);
        }

通过调用isInterrupted（）可以判断实例线程是否被中断。

它的内部调用了isInterrupted(false)方法：

/** * Tests if some Thread has been interrupted. The interrupted state * is reset or not based on the value of ClearInterrupted that is * passed. */
        private native boolean isInterrupted(boolean ClearInterrupted);

这个方法是个native方法，接收一个是否清除Interrupted标志位的参数。

我们可以看到isInterrupted()传入的参数是false，这就表示isInterrupted()只会判断是否被中断，而不会清除中断状态。

## interrupted ##

interrupted是Thread中的一个类方法：

public static boolean interrupted() { 
            return currentThread().isInterrupted(true);
        }

我们可以看到，interrupted（）也调用了isInterrupted(true)方法，不过它传递的参数是true，表示将会清除中断标志位。

> 注意，因为interrupted()是一个类方法，调用isInterrupted(true)判断的是当前线程是否被中断。注意这里currentThread（）的使用。

## interrupt ##

前面两个是判断是否中断的方法，而interrupt（）就是真正触发中断的方法。

我们先看下interrupt的定义：

public void interrupt() { 
            if (this != Thread.currentThread())
                checkAccess();
    
            synchronized (blockerLock) { 
                Interruptible b = blocker;
                if (b != null) { 
                    interrupt0();           // Just to set the interrupt flag
                    b.interrupt(this);
                    return;
                }
            }
            interrupt0();
        }

从定义我们可以看到interrupt（）是一个实例方法。

它的工作要点有下面4点：

1.  如果当前线程实例在调用Object类的wait（），wait（long）或wait（long，int）方法或join（），join（long），join（long，int）方法，或者在该实例中调用了Thread.sleep（long）或Thread.sleep（long，int）方法，并且正在阻塞状态中时，则其中断状态将被清除，并将收到InterruptedException。
2.  如果此线程在InterruptibleChannel上的I / O操作中处于被阻塞状态，则该channel将被关闭，该线程的中断状态将被设置为true，并且该线程将收到java.nio.channels.ClosedByInterruptException异常。
3.  如果此线程在java.nio.channels.Selector中处于被被阻塞状态，则将设置该线程的中断状态为true，并且它将立即从select操作中返回。
4.  如果上面的情况都不成立，则设置中断状态为true。

我们来举个例子：

@Slf4j
 public class InterruptThread extends Thread { 
 @Override
 public void run() { 
 for (int i = 0; i < 1000; i++) { 
 log.info("i= {}", (i+1));
 log.info("call inside thread.interrupted()： {}", Thread.interrupted());
 }
 }
 
 @Test
 public void testInterrupt(){ 
 InterruptThread thread=new InterruptThread();
 thread.start();
 thread.interrupt();
 //test isInterrupted
 log.info("first call isInterrupted(): {}", thread.isInterrupted());
 log.info("second call isInterrupted(): {}", thread.isInterrupted());
 
 //test interrupted（)
 log.info("first call outside thread.interrupted()： {}", Thread.interrupted());
 log.info("second call outside thread.interrupted() {}：", Thread.interrupted());
 log.info("thread is alive : {}",thread.isAlive() );
 }
 }

输出结果如下：

13:07:17.804 [main] INFO com.flydean.InterruptThread - first call isInterrupted(): true
    13:07:17.808 [main] INFO com.flydean.InterruptThread - second call isInterrupted(): true
    
    13:07:17.808 [Thread-1] INFO com.flydean.InterruptThread - call inside thread.interrupted()： true
    13:07:17.808 [Thread-1] INFO com.flydean.InterruptThread - call inside thread.interrupted()： false
    
    13:07:17.808 [main] INFO com.flydean.InterruptThread - first call outside thread.interrupted()： false
    13:07:17.809 [main] INFO com.flydean.InterruptThread - second call outside thread.interrupted() false

上面的例子中，两次调用thread.isInterrupted()的值都是true。

在线程内部调用Thread.interrupted()， 只有第一次返回的是ture，后面返回的都是false，这表明第一次被重置了。

在线程外部，因为并没有中断外部线程，所以返回的值一直都是false。

# 总结 #

本文介绍了java并发系列文章1到14章，因为文件篇幅限制，剩下的章节将会在  
5W字高质量java并发系列详解教程(下) 进行介绍，敬请期待！

本文的例子[https://github.com/ddean2009/learn-java-concurrency/][https_github.com_ddean2009_learn-java-concurrency]

本文PDF下载链接[concurrent-all-in-one.pdf][]

> 欢迎关注我的公众号:程序那些事，更多精彩等着您！  
> 更多内容请访问 www.flydean.com

[concurrent-all-in-one.pdf]: https://github.com/ddean2009/www.flydean.com/blob/master/java/concurrent/concurrent-all-in-one.pdf
[https_github.com_ddean2009_learn-java-concurrency]: https://github.com/ddean2009/learn-java-concurrency/
[eb7ae88ab10b06c14e3017c2799038bf.png]: /images/20220829/dd35ba0294534b25b02399637c2cfabe.png