JUC并发编程：第二季

1、用户线程和守护线程

package com.mace.juc2;

public class DemonThread {
	public static void main(String[] args) {
		//This method must be invoked before the thread is started.
		//setDaemon()设置为守护线程时必须在线程启动之前，
		//否则：IllegalThreadStateException非法的线程状态异常
		boolean daemon = Thread.currentThread().isDaemon();
		System.out.println(daemon);
	}
}

2、CompletableFuture异步任务

Future 定义了对异步任务的操作，例如：取消异步任务、获取结果、判断任务是否结束等。

public class FeatureDemo {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		FutureTask futureTask = new FutureTask<>(new Mythread());
		Thread thread = new Thread(futureTask);
		thread.start();
		String string = futureTask.get();
		System.out.println(string);
	}
}
class Mythread implements Callable{
	@Override
	public String call() throws Exception {
		// TODO Auto-generated method stub
		return "Hello call() "+Thread.currentThread().getName();
	}
}

public class FutureTaskDemo {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		FutureTask task = new FutureTask<>(()->{
			TimeUnit.SECONDS.sleep(5);
			return "TASK OVER";
		});
		
		Thread thread = new Thread(task);
		
		thread.start();
		
		System.out.println("Main thread over");
		
		while(true) {
			if(task.isDone()) {
				System.out.println(task.get());
				break;
			}else {
				TimeUnit.MILLISECONDS.sleep(500);
				System.out.println("任务未结束，请等待...");
			}
		}
	}
}

CompletableFuture提供了一种观察者类似的模式，可以在任务完成后通知监听的一方。
public class CompletableFuture implements Future, CompletionStage
CompletionStage：代表异步计算过程中的某个阶段，一个阶段完成后可能会触发另外一个阶段。
四个常用静态方法：

public static CompletableFuture runAsync(Runnable runnable):无返回值
public static CompletableFuture supplyAsync(Supplier supplier):有返回值
可以指定线程池，否则使用ForkJoinPool。

public class CompleteablefutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //使用默认的线程池ForkJoinPool.commonPool，无返回值。
        CompletableFuture v1 = CompletableFuture.runAsync(()->{
            System.out.println("Hello runAsync");
            System.out.println(Thread.currentThread().getName());
        });
        System.out.println(v1.get());

        ExecutorService pool = Executors.newFixedThreadPool(3);
        //有返回值且使用自己创建的线程池
        CompletableFuture v2 = CompletableFuture.supplyAsync(()->{
            System.out.println("Hello supplyAsync");
            System.out.println(Thread.currentThread().getName());
            return "Hello World";
        },pool);
        System.out.println(v2.get());
        pool.shutdown();
    }
}

2.1 CompletableFuture增强Future语法：

public class CompletableFuturePro {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //使用默认的线程池，相当于守护线程，主线程结束后自动结束得不到结果。
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " come in");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int i = ThreadLocalRandom.current().nextInt(10);
            System.out.println("计算出结果：" + i);
            if(i<4){
                int x = i/0;
            }
            return i;
        },executorService).whenComplete((x,e)->{//完成时调用
            if(e == null){
                System.out.println("获取到结果："+x);
            }
        }).exceptionally(e->{//出现异常时调用
            e.printStackTrace();
            System.out.println(e.getCause()+"t"+e.getMessage());
            return null;
        });

        System.out.println(Thread.currentThread().getName()+" do other something");
    }

    private static void taskone() throws InterruptedException, ExecutionException {
        CompletableFuture task = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " come in");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int i = ThreadLocalRandom.current().nextInt(10);
            System.out.println("计算出结果：" + i);
            return i;
        });

        System.out.println(Thread.currentThread().getName()+" do other something");

        System.out.println(task.get());
    }
}

CompletableFuture流式计算，get方法有异常，join方法没有异常。

package com.mace.juc2;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;



public class CompleteableFutureDemo {
	
	public static List malls = Arrays.asList(
			new NetMall("jd"),new NetMall("dd"),new NetMall("tb")
	);
	
	//step by step
	public static List StepByStep(List malls,String name){
		return malls.stream().map((n)->{
			return String.format("《 %s 》 in %s price is %.2f", name,n.getName(),n.caleMall(name));
		}).collect(Collectors.toList());
	}
	//CompletableFuture异步任务实现
	public static List FutureSalce(List malls,String name){
		//map映射后必须进行collect收集计算，否则不是异步执行。
		return malls.stream().map(mall -> {
			return CompletableFuture.supplyAsync(()-> {
				return String.format("《 %s 》 in %s price is %.2f", name,mall.getName(),mall.caleMall(name));
			}
			);
		}).collect(Collectors.toList()).stream().map(f -> {
			return f.join();
		}).collect(Collectors.toList());
	}
	
	public static void main(String[] args) {
		long start = System.currentTimeMillis();
		List stepByStep = StepByStep(malls,"mysql");
		System.out.println(stepByStep.toString());
		long end = System.currentTimeMillis();
		System.out.println("耗时："+(end-start)+"毫秒");//3513
		
		System.out.println("-----------------");
		
		long start2 = System.currentTimeMillis();
		List futureSalce = FutureSalce(malls,"mysql");
		System.out.println(futureSalce.toString());
		long end2 = System.currentTimeMillis();
		System.out.println("耗时："+(end2-start2)+"毫秒");//1022
	}
}


class NetMall{
	
	private String name;
	
	public NetMall(String name) {
		super();
		this.name = name;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}
	
	public double caleMall(String name) {
		double cale =ThreadLocalRandom.current().nextDouble()*2+name.charAt(0);
		try {
			TimeUnit.SECONDS.sleep(1);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return cale;
	}
}

2.2 CompletableFuture常用API:

获得结果和触发计算

get() 不见不散
get(time) 过时不候
join()
getNow(valueifabsent) 立即获取结果，任务完毕则正常返回，否则返回默认值。
complete(value) 主动触发计算，是否打断get方法返回默认值。

public class ComAbleFutureAPI {
	public static void main(String[] args) throws Exception{
		CompletableFuture task = CompletableFuture.supplyAsync(()->{
			try {
				TimeUnit.SECONDS.sleep(2);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			return "Java";
		});
		
		//延时等待任务完成，则正常获取计算结果。
		try {
			TimeUnit.SECONDS.sleep(3);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		//直接获取，任务没有计算完成，返回默认值。
		boolean complete = task.complete("python");
		System.out.println(task.get());
	}
	
}

对计算结果进行处理

thenApply 计算结果存在依赖关系，两个线程串行化。有异常直接停止。
handle 计算结果存在依赖关系，两个线程串行化。有异常也可以往下处理。

public class ComAbleFutureAPI2 {
	public static void main(String[] args) throws Exception{
		CompletableFuture.supplyAsync(()->{
			System.out.println("one step");
			return 1;
		}).thenApply(v->{
			System.out.println("two step");
			return v + 1;
		}).thenApply(v->{
			System.out.println("three step");
			return v + 2;
		}).whenComplete((v,e)->{
			if(e== null) {
				System.out.println(v);
			}
		}).exceptionally(e->{
			e.printStackTrace();
			return null;
		});
	}
}

对计算结果进行消费

thenAccept 接受任务处理的结果，并消费处理，无返回值。

thenrun和thenrunAsync的区别：

对计算速度进行选用
applyToEither 选用最快的

public class ComAbleFutureAPI3 {
	public static void main(String[] args) throws Exception{
		CompletableFuture t1 = CompletableFuture.supplyAsync(()->{
			try {
				TimeUnit.SECONDS.sleep(3);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			return "A";
		});
		
		CompletableFuture t2 = CompletableFuture.supplyAsync(()->{
			try {
				TimeUnit.SECONDS.sleep(2);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			return "B";
		});
		
		CompletableFuture applyToEither = t1.applyToEither(t2, (f)->{
			return f + " is fast";
		});
		
		System.out.println(applyToEither.join());
		
		System.out.println(Thread.currentThread().getName()+" do something。。。");
	}
}

对计算结果进行合并

thenCombine 对任务的结果合并处理

3、Java锁 3.1 乐观锁和悲观锁

悲观锁：synchronized关键字和lock的实现类都是悲观锁，认为自己在使用数据的时候一定有别的线程来修改数据，因此在获取数据的时候会加锁，确保数据不会被别的线程修改。
乐观锁：认为自己在使用数据时不会有别的线程来修改数据或资源，不会添加锁。在Java中通过无锁编程来实现，只是在更新数据前判断，之前有没有别的线程更新了这个数据，如果没有被更新正常写入，如果已经被更新了则放弃操作或者其他处理。实现方式：version版本号；CAS算法。

3.2 对象锁和类锁

对象锁：普通同步方法用的都是同一把锁-new出来的具体实例对象this，当一个同步方法被调用时，锁的是整个资源类，并不是某一个方法，其他的同步方法必须等锁被释放后才能获取锁。
类锁：静态同步方法用的是同一把锁-类对象本身即模板class,与对象锁独立，只有锁被释放后其他的静态同步方法才能获取锁。

class Phone{
	
	public synchronized void sendEmail() {
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.out.println("send email");
	}
	
	public static synchronized void sendEms() {
		System.out.println("send ems");
	}
	
}


public class 对象锁和类锁 {
	public static void main(String[] args) {
		Phone phone = new Phone();
		
		new Thread(()->{
			phone.sendEmail();	
		}).start();
		
		try {
			TimeUnit.SECONDS.sleep(1);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		new Thread(()->{
			phone.sendEms();	
		}).start();
	}
}

3.3 公平锁和非公平锁

公平锁：多个线程按照申请锁的顺序获取锁
非公平锁：多个线程不是按照申请锁的顺序获取锁，后申请的可能先获得锁，在高并发下，有可能造成优先级翻转或者饥饿的状态。

package com.mace.juc2;


import java.util.concurrent.locks.ReentrantLock;

class Tickets{
	
	private int ticket = 50;
	
	//ReentrantLock lock = new ReentrantLock();//默认非公平相当于传入false
	ReentrantLock lock = new ReentrantLock(true);//公平锁
	
	public void sale() {
		lock.lock();
		try {
			if(ticket > 0) {
				System.out.println(Thread.currentThread().getName()+" sale "+ ticket +" ticket");
				ticket--;
			}
		} catch (Exception e) {
			// TODO: handle exception
		}finally {
			lock.unlock();
		}
	}
}



public class FairLock {
	public static void main(String[] args) {
		Tickets tickets = new Tickets();
		new Thread(()->{
			for (int i = 0; i < 60; i++) {
				tickets.sale();
			}
		},"A").start();
		
		new Thread(()->{
			for (int i = 0; i < 60; i++) {
				tickets.sale();
			}
		},"B").start();
		
		new Thread(()->{
			for (int i = 0; i < 60; i++) {
				tickets.sale();
			}
		},"C").start();
	}
}

3.4 可重入锁（递归锁）

指同一个线程在外层方法获取锁的时候，再进入该线程的内部方法会自动获取锁（锁对象是同一把锁），不会因为之前已经获取过没有释放而阻塞。ReentrantLock及synchronized都是可重入锁，可重入锁的优点是可一定程度避免死锁。

隐式锁：synchronized使用的锁

public class SyncReenLock {
	public static void main(String[] args) {
		Object lock = new Object();
		new Thread(()->{
			synchronized (lock) {
				System.out.println(Thread.currentThread().getName()+" come in 外层");
				synchronized (lock) {
					System.out.println(Thread.currentThread().getName()+" come in 中层");
					synchronized (lock) {
						System.out.println(Thread.currentThread().getName()+" come in 内层");
					}
				}
			}
		}).start();
	}
}

显式锁：ReentrantLock加锁和释放锁必须一一匹配，否则将出阻塞或者死锁。

3.5 死锁

3.5.1死锁的排查

方式一：

jps -l 查看进程编号
jstack 编号
方式二：
jconsole图形化界面

4、LockSupport与中断机制 4.1、LockSupport

概念：用于创建锁和其他同步类的基本线程阻塞原语。
该类与使用它的每个线程关联一个许可证（在Semaphore类的意义上）。如果许可证可用，将立即返回park ，并在此过程中消费; 否则可能会阻止。如果尚未提供许可，则致电unpark获得许可。（与Semaphores不同，许可证不会累积。最多只有一个。）可靠的使用需要使用volatile（或原子）变量来控制何时停放或取消停放。对于易失性变量访问保持对这些方法的调用的顺序，但不一定是非易失性变量访问。

4.1.1 线程等待唤醒机制

Object类的wait及notify
Condition接口的await()和signal()
LockSupport类的park和unpark

4.2、中断机制

概念：一个线程不应该由其他线程来强制中断或停止，而是应该由该线程自己停止，在Java中没有办法立即停止一条线程，Java提供了一种用于停止线程的协商机制-中断，也即中断标识协商机制。中断的过程完全需要程序员自己实现，若要中断一个线程，需要手动调用interrupt方法，该方法也仅仅将线程对象的中断标识设为true，后续判断中断标识等处理需手动实现。

public void interrupt() 设置线程的中断状态为true
public static boolean interrupted() 测试当前线程是否已被中断。此方法清除线程的中断状态。
public boolean isInterrupted() 测试此线程是否已被中断

4.2.2 停止中断运行中的线程

public class StopThreadByVoliate {
    //线程之间可见
    private static volatile boolean isStop = false;
    public static void main(String[] args) throws Exception {
        new Thread(()->{
            while (true){
                if(isStop){
                    System.out.println("isStop="+isStop);
                    break;
                }
                System.out.println("thread running...");
            }
        },"a").start();

        TimeUnit.SECONDS.sleep(1);
        isStop = true;
    }
}

public class StopThreadByAtomicBoolean {
    static AtomicBoolean flag = new AtomicBoolean(false);
    public static void main(String[] args) throws InterruptedException {
        new Thread(()->{
            while (true){
                if(flag.get()){
                    System.out.println("flag="+flag);
                    break;
                }
                System.out.println("thread running...");
            }
        },"a").start();

        TimeUnit.MILLISECONDS.sleep(1);
        flag.set(true);
    }
}

public class StopThreadByInterrupt {
    public static void main(String[] args) throws InterruptedException {
        Thread t = new Thread(()->{
            while (true){
                if(Thread.currentThread().isInterrupted()){
                    System.out.println("isInterrupted="+Thread.currentThread().isInterrupted());
                    break;
                }
                System.out.println("thread running...");
            }
        },"a");

        t.start();

        System.out.println("默认："+t.isInterrupted());

        TimeUnit.MILLISECONDS.sleep(1);
        t.interrupt();
    }
}

public class TestInterruptException {
    public static void main(String[] args) throws Exception {
        Thread t = new Thread(() -> {
            while (true) {
                if(Thread.currentThread().isInterrupted()){
                    System.out.println("thread is interrupted ");
                    break;
                }

                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    //如果该线程阻塞的调用wait() ， wait(long) ，或wait(long, int)的方法Object类，
                    // 或的join() ， join(long) ， join(long, int) ， sleep(long) ，或sleep(long, int) ，
                    // 这个类的方法，那么它的中断状态将被清除，并且将收到InterrupsstedException 。
                    //清除后需要重新设置
                    Thread.currentThread().interrupt();
                    e.printStackTrace();
                }

                System.out.println("thread running...");
            }
        });

        t.start();

        TimeUnit.SECONDS.sleep(1);

        new Thread(()->{
            t.interrupt();
        }).start();
    }
}

总结：

JUC并发编程：第二季

Java相关栏目本月热门文章