并发编程--上篇 - qidao123.com技术社区-IT企服评测·应用市场

public enum State {
// 尚未启动的线程的线程状态。
NEW,
// 就绪
RUNNABLE,
// 等待监视器锁的线程的线程状态
BLOCKED,
/*
等待线程的线程状态,线程由于调用以下方法之一而处于等待状态：
Object.wait() 没有超时
Thread.join() 没有超时
LockSupport.park()
*/
WAITING,
/*
指定等待时间的等待线程的线程状态
线程处于定时等待状态，因为调用了以下方法之一，并指定等待时间：
Thread.sleep
Object.wait with timeout
Thread.join with timeout
LockSupport.parkNanos
LockSupport.parkUntil
*/
TIMED_WAITING,
//终止线程的线程状态。线程已完成执行。
TERMINATED;
}

复制代码

public class MyThread1 extends Thread {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + ": hello world");
}
}
public class JUCMain {
public static void main(String[] args) {
new MyThread1().start();
}
}

复制代码

public class Runnable1 implements Runnable{
@Override
public void run() {
System.out.println("hello world, Runnable");
}
}
public class JUCMain {
public static void main(String[] args) {
new Thread(new Runnable1()).start();
}
}

复制代码

// 启动线程并触发 JVM 创建原生线程
// synchronized后面解释【见探索“锁”】
public synchronized void start() {
// 零状态值对应于状态 “NEW”
// 线程想要start，必须是为0的状态
if (threadStatus != 0)
throw new IllegalThreadStateException();
/*
group 是线程所属的线程组。这行代码将当前线程实例添加到线程组中，
同时线程组的未启动线程计数会减1。
*/
group.add(this);
boolean started = false;
try {
start0(); //关键！调用本地方法（native）
started = true;
} finally {
try {
if (!started) { //启动失败时回滚
//如果 started 为 false，说明线程启动失败，
//调用 group.threadStartFailed(this) 方法通知线程组该线程启动失败。
group.threadStartFailed(this);
}
} catch (Throwable ignore) {
/* do nothing. If start0 threw a Throwable then
it will be passed up the call stack */
}
}
}
//========== native
private native void start0();

复制代码

private Runnable target; // target是Runnable类型
@Override
public void run() {
if (target != null) {
target.run();
}
}

复制代码

@FunctionalInterface
public interface Runnable {
public abstract void run();
}

复制代码

public class MyCall implements Callable<String> {
@Override
public String call() throws Exception {
Thread.sleep(2000);
return "Hello Callable";
}
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutureTask<String> task = new FutureTask<>(new MyCall());
new Thread(task).start();
System.out.println(task.get());
}

复制代码

// 这里使用了泛型 <V>，表示该 Future 对象所代表的异步计算结果的类型。
public interface Future<V> {
//尝试取消异步任务的执行。
/*
如果任务已经完成、已经被取消或者由于其他原因无法取消，则返回 false；
如果任务成功取消，则返回 true。
*/
boolean cancel(boolean mayInterruptIfRunning);
//如果任务在完成之前被取消，则返回 true；否则返回 false。
boolean isCancelled();
//如果任务已经完成，则返回 true；否则返回 false。
boolean isDone();
//获取异步任务的计算结果。如果任务还未完成，调用该方法的线程会被阻塞，直到任务完成。
V get() throws InterruptedException, ExecutionException;
//获取异步任务的计算结果，并且可以指定一个超时时间。
//如果在指定的时间内任务还未完成，调用该方法的线程会被阻塞，直到任务完成或者超时。
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}

复制代码

public interface RunnableFuture<V> extends Runnable, Future<V> {
// 很简单嘛，这个是来自Runnable的
void run();
}

复制代码

public class FutureTask<V> implements RunnableFuture<V> {
// 基本属性
private volatile int state;
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** 结果 */
private Object outcome;
/** The thread running the callable; CAS ed during run() */
private volatile Thread runner;
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;
// 看它的构造函数1
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable; // 赋值callable========
this.state = NEW; // ensure visibility of callable
}
// 构造函数2 ==== 本质还是把Runnable加了一层，给封装成Callable了
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
/*
Executors::callable(xx, xx)方法==========
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run(); // 调用Runnable的run()
return result;
}
}
*/
// run()方法 ---------------
// new Thread(new FutureTask<>(new MyCall()))
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//====调用callable.call()
result = c.call();
ran = true;
} catch (Throwable ex) {
.........
}
// 如果运行OK了，设置结果！
if (ran) set(result);
}
} finally {
.............
}
}
// 设置结果outcome
protected void set(V v) {
// https://www.cnblogs.com/jackjavacpp/p/18787832
// 使用CAS --- 【见上一篇文章 java map & CAS & AQS】
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v; // 这里
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
// 比较核心的get方法================start
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING) // 如果状态不是完成
s = awaitDone(false, 0L); // 等待完成
return report(s); // 返回结果
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 1.计算超时截止时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) { // 2.自旋循环等待任务完成
// 2.1如果该线程中断了
if (Thread.interrupted()) {
removeWaiter(q);// 从等待队列中移除当前节点
throw new InterruptedException();
}
// 2.2检查状态
int s = state;
// 任务已终态（NORMAL, EXCEPTIONAL, CANCELLED）
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;// 返回最终状态
}
// 2.3若任务状态等于 COMPLETING，表明任务正在完成，
// 此时调用 Thread.yield() 方法让当前线程让出 CPU 时间片，等待任务完成。
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued) //将节点加入等待队列
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) { // 2.4如果是有时限的get()
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state; // 返回状态
}
LockSupport.parkNanos(this, nanos);
}
else //若没有设置超时时间，就调用 LockSupport.park 方法让当前线程无限期阻塞，直到被唤醒。
LockSupport.park(this);
}
}
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x; // 返回outcome
......
}
//==================================end
}

复制代码

public class PoolMain {
public static void main(String[] args) {
// 创建一个线程池
ExecutorService pool = Executors.newFixedThreadPool(1);
long start = System.currentTimeMillis();
// execute=============
pool.execute(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("execute pool创建启动线程!");
});
// submit==============
Future<Integer> future = pool.submit(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("submit pool创建启动线程!");
return 100;
});
try {
System.out.println(future.get());
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException(e);
}
System.out.println("main线程执行时间：" + (System.currentTimeMillis() - start));
pool.shutdown();
}
}

复制代码