Java多线程、并发

    并发编程使我们可以将程序划分为多个分离的、独立运行的任务。通过多线程机制，这些独立的任务中的每一个都将由执行线程来驱动。单个进程可以拥有多个并发执行任务。
    实现并发最直接的方式是在操作系统级别使用进程。
    Java的线程机制是抢占式的，调度机制会周期性的中断线程，将上下文切换到另一个线程，从而为每个线程都提供时间片，使得每个线程都会分配到数量合理的赶时间去驱动它的任务。还有一种方式是协作式，这种需要程序主动去释放线程控制权。

一、多线程使用

1、继承Runnable接口

 1 public class MyRunnable implements Runnable {
 2     private int index;
 3     @Override
 4     public void run() {
 5         while (index ++ < 10){
 6             System.out.print(index + ",");
 7             Thread.yield();//可选使用，用来告诉线程调试器，当前任务已执行完，可以切换线程了
 8         }
 9     }
10     public static void main(String[] args) {
11         //启用新线程
12         new Thread(new MyRunnable()).start();
13         new Thread(new MyRunnable()).start();
14         System.out.println("main ended");
15     }
16 }

结果：

main ended
1,2,1,2,3,3,4,4,5,5,6,6,7,7,8,8,9,9,10,10,

相关参数设置：

    Thread thread1 = new Thread(new MyRunnable());
    //设置thread1为后台线程，程序会在所有非后台线程运行结束后结束，而不关心后台线程是否已经结束了。
    thread1.setDaemon(true);
    thread1.setName("threadname");//设置线程名称
    thread1.start();

    Thread thread2 = new Thread(new MyRunnable());
    thread2.start();
    try {
        thread2.join(6000); //主线程会在此等待thread2执行完，最多等待6秒，如果不设置超时时间则一直等待
        thread2.interrupt();//取消等待，这个与join在同一个线程执行是没有效果的。
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

使用Executors来管理线程：

    //SE5后可以使用Executors来管理线程
    ExecutorService pool = Executors.newCachedThreadPool();
    //相似的还有FixedThreadPool(固定线程)， SingleThreadExecutor单线程（如果有多个任务会排队）
//  Executors.newFixedThreadPool(10);
//  Executors.newSingleThreadExecutor();
    pool.execute(new MyRunnable());
    pool.shutdown();//shutdown后就不能添加新的线程任务了

使用Executors来执行线程还可以获取到线程内的异常getUncaughtExceptionHandler；   

2、继承Callable接口

可获取线程任务返回值

 1 public class MyCallable implements Callable<String> {
 2     private int index;
 3     @Override
 4     public String call() throws Exception {
 5         while (index ++ < 10){
 6             System.out.print(index + ",");
 7         }
 8         return "success";
 9     }
10 
11     public static void main(String[] args) throws ExecutionException, InterruptedException {
12         ExecutorService pool = Executors.newCachedThreadPool();
13         Future<String> result = pool.submit(new MyCallable());
14         result.isDone();  //返回当前任务是否已经完成，未完成时调用get()会一直阻塞，直到完成返回结果为止
15         System.out.println(result.get()); //success
16     }
17 }

二、共享资源加锁

1、synchronized关键字

可使用synchronized关键字来修饰方法，防止资源冲突。
如：synchronized void f(){}
    synchronized void g(){}
当在对象上调用其任意synchronized方法时，此对象都会被加锁，此时该对象上的其他synchronized方法只有等前一个方法调用完毕释放锁之后才能被调用。
以上如某个任务对对象调用了f(),对于同一个对象而言，就只能等f()调用结束并释放锁之后其他任务才能调用f()和g()。所以对于某个特定对象来说synchronized共享同一个锁。
但同一个任务可多次获得对象锁，如果f()调用了g()，也是可以的，这时锁的计数就是2，只有当锁的计数变为0时，锁才会被释放。
注意在使用并发时，将被共享资源的域设置为private是非常重要的，否则synchronized关键字就不能防止其他任务直接访问域。
示例：

 1 //多线程计数
 2 public class SynchObject {
 3     private Integer index = 0 ;
 4     synchronized public void counting(){
 5         while (index ++ < 10){
 6             System.out.printf(index + ",");
 7             try {
 8                 Thread.sleep(100L);
 9             } catch (InterruptedException e) {
10                 e.printStackTrace();
11             }
12         }
13     }
14 }
15 public class SyncTesthRunnable implements Runnable {
16     SynchObject synchObject;
17     SyncTesthRunnable(SynchObject synchObject){
18         this.synchObject = synchObject;
19     }
20     @Override
21     public void run() {
22     
23     
24         this.synchObject.counting();
25     }
26     public static void main(String[] args){
27         SynchObject synchObject = new SynchObject();
28         new Thread(new SyncTesthRunnable(synchObject)).start();
29         new Thread(new SyncTesthRunnable(synchObject)).start();
30     }
31 }

如果counting方法不加synchronized关键字，计数的结果就不会是顺序的；
不加synchronized时的运行结果：2,2,3,3,4,5,7,7,8,9,10,
加synchronized时的运行结果：1,2,3,4,5,6,7,8,9,10,

2、显示的使用锁Lock对象来加锁

使用如下：

 1 public class SynchObject {
 2     private Integer index = 0 ;
 3     private Lock lock = new ReentrantLock();
 4 
 5     public void counting(){
 6         lock.lock();
 7         while (index ++ < 10){
 8             System.out.printf(index + ",");
 9             try {
10                 Thread.sleep(100L);
11             } catch (InterruptedException e) {
12                 e.printStackTrace();
13             }
14         }
15         lock.unlock();//注意锁一定要主动释放，不然其他任务就永远获取不到锁。最好能放在finally里，以防异常发生无法释放锁。
16     }
17 }

如果在synchronized里事务失败了则会抛出异常，我们没有机会去做一些清理工作。使用Lock锁则可以做更多的动作。通常使用synchronized就足够了，只有在解决特殊问题时才显示使用Lock对象。
*在类中只有一个可变域，且对域只进行原子操作时，可使用volatile关键字来限定域，确保同步。但最好不要用，很可能因为理解不到位出现同步问题，一般都推荐直接使用synchronized。
*什么是原子操作？对域中(除long,double外的基本类型)的值做赋值和返回值操作通常都是原子性的。如果域有递增递减操作的肯定不是原子操作。

3、临界区

只对方法内的部分代码加锁

    synchronized(synchObject){
        //这个代码块里的代码只能同时被一个线程访问，也称为同步控制块
    }

三、中断线程

ExecutorService exec = Executors.newCachedThreadPool();
Future<String> future = exec.submit(new MyCallable());
future.cancel(true);//方式一
exec.shutdownNow();//方式二
Thread.interrupted();// 检查当前线程是否中断

四、线程间协作

1、wait(),notifyAll()/notify()

 1 public class SynchObject {
 2     private boolean waked = false;
 3 
 4     public synchronized void wakeup() throws InterruptedException {
 5         Thread.sleep(1000L);//sleep()和yield()方法不会释放锁, wait()会释放锁
 6         waked = true;
 7         System.out.println("wake up");
 8         notifyAll(); //notifyAll()唤醒所有的等待，notify()只唤醒一个。如果确定只有一个可以使用notify来进行优化，其他情况建议使用notifyAll
 9     }
10     public synchronized void getup() throws InterruptedException {
11         while (!waked){//wait一般注意要加判断条件，因为不能确定哪个线程先执行，不判断可能造成选notify了再wait，那么就可能一直等不到通知，造成线程一直挂起
12             System.out.println("waiting");
13             wait(); //wait和notify方法都需要使用synchronized锁定对象线程，否则会报IllegalMonitorStateException异常
14         }
15         System.out.println("get up");
16     }
17 
18     public class WakeupTask implements Runnable{
19         private SynchObject object;
20         WakeupTask(SynchObject object){this.object = object;}
21         public void run() {
22             try {
23                 this.object.wakeup();
24             } catch (InterruptedException e) {
25                 e.printStackTrace();
26             }
27         }
28     }
29     public class GetupTask implements Runnable{
30         private SynchObject object;
31         GetupTask(SynchObject object){this.object = object;}
32         public void run() {
33             try {
34                 this.object.getup();
35             } catch (InterruptedException e) {
36                 e.printStackTrace();
37             }
38         }
39     }
40 
41     public void test() {
42         SynchObject object = new SynchObject();
43         WakeupTask wakeupTask = new WakeupTask(object);
44         GetupTask getupTask = new GetupTask(object);
45         new Thread(getupTask).start();
46         new Thread(wakeupTask).start();
47     }
48 }
-------
test()运行结果：
waiting
wake up
get up

2、生产-消费者与队列

更高效的协作

 1 class Person{
 2     private String name;
 3     Person(String name){this.name = name; }
 4     void wakeup(){System.out.println(name + " wake up");}
 5     void getup() throws InterruptedException {
 6         Thread.sleep((long) (1000L * Math.random()));
 7         System.out.println(name + "get up");
 8     }
 9     void working(){System.out.println(name + " working");}
10 }
11 class PersonQueue extends LinkedBlockingQueue<Person>{}
12 
13 class WakedPerson implements Runnable{
14     private PersonQueue personQueue;
15     public WakedPerson(PersonQueue personQueue){this.personQueue = personQueue;}
16 
17     public void run() {
18         for (Integer i = 0; i< 5; i++){
19             Person p = new Person(i.toString());
20             p.wakeup();
21             personQueue.add(p);//也可用put()
22         }
23     }
24 }
25 class GetupPerson implements Runnable{
26     private PersonQueue wakedQueue;
27     private PersonQueue gotupQueue;
28     GetupPerson (PersonQueue personQueue, PersonQueue gotupQueue){
29         this.wakedQueue = personQueue;
30         this.gotupQueue = gotupQueue;
31     }
32     public void run(){
33         try {
34             while (!Thread.interrupted()){
35                 Person p = wakedQueue.take();
36                 p.getup();
37                 gotupQueue.put(p);
38             }
39         } catch (InterruptedException e) {
40             System.out.println("GetupPerson thread ended");
41         }
42     }
43 }
44 class WorkingPerson implements Runnable{
45     private PersonQueue gotupQueue;
46     WorkingPerson(PersonQueue gotupQueue){this.gotupQueue = gotupQueue; }
47     public void run(){
48         try {
49             while (!Thread.interrupted()){
50                 Person p = gotupQueue.take();
51                 p.working();
52             }
53         } catch (InterruptedException e){
54             System.out.println("WorkingPerson thread ended");
55         }
56     }
57 }
58 
59 void test() throws InterruptedException {
60     PersonQueue personQueue = new PersonQueue();
61     PersonQueue gotupQueue = new PersonQueue();
62     ExecutorService exec = Executors.newCachedThreadPool();
63     exec.execute(new WakedPerson(personQueue));
64     exec.execute(new GetupPerson(personQueue, gotupQueue));
65     exec.execute(new WorkingPerson(gotupQueue));
66     //后两个线程的写法不会主动结束，这里有需要的话可以主动结束
67     TimeUnit.SECONDS.sleep(5L);
68     exec.shutdownNow();
69 }
----------
test()运行结果：
0 wake up
1 wake up
2 wake up
3 wake up
0get up
0 working
1get up
1 working
2get up
2 working
3get up
3 working
GetupPerson thread ended
WorkingPerson thread ended
test()运行结果：

3、管道通信

另一种协作，不同平台之间可能存在差异，相对而言BlockingQueue更健壮

 1 class Sender implements Runnable{
 2     private Random random = new Random(46);
 3     private PipedWriter writer = new PipedWriter();
 4     public PipedWriter getWriter(){return writer;}
 5     public void run() {
 6         for (char i = 'A'; i < 'G'; i++){
 7             try {
 8                 writer.write(i);
 9                 TimeUnit.MILLISECONDS.sleep(random.nextInt(500));
10             } catch (IOException | InterruptedException e) {
11                 e.printStackTrace();
12             }
13         }
14     }
15 }
16 
17 class Receiver implements Runnable{
18     private PipedReader reader;
19     Receiver(Sender sender) throws IOException {
20         this.reader = new PipedReader(sender.getWriter());
21     }
22     public void run(){
23         while (true){
24             try {
25                 System.out.println("read: " + (char)reader.read());
26             } catch (IOException e) {
27                 e.printStackTrace();
28             }
29         }
30     }
31 }
32 void test() throws IOException {
33     Sender sender = new Sender();
34     Receiver receiver = new Receiver(sender);
35     ExecutorService excu = Executors.newCachedThreadPool();
36     excu.execute(receiver);
37     excu.execute(sender);
38 }
----------
test()运行结果：
read: A
read: B
read: C
read: D
read: E
read: F
test()运行结果

五、SE5引入的线程处理构件(类)

1、同步一个或多个任务，强制它们等待另一组任务的执行完成（CountDownLatch）

 1 class PreTask implements Runnable{
 2     private final Integer index;
 3     private CyclicBarrier cyclicBarrier;
 4     PreTask(Integer index, CyclicBarrier cyclicBarrier) {
 5         this.index = index;
 6         this.cyclicBarrier = cyclicBarrier;
 7     }
 8     public void run() {
 9         try {
10             TimeUnit.MILLISECONDS.sleep(100 * index);
11             cyclicBarrier.await();
12             System.out.print(index + ",");
13         } catch (InterruptedException | BrokenBarrierException e) {}
14     }
15 }
16 
17 class AfterTask{
18     private ExecutorService excu = Executors.newCachedThreadPool();
19     private CyclicBarrier barrier;
20     private int index = 0;
21 
22     public void runn(){
23         barrier = new CyclicBarrier(3, new Runnable() {//3表示cycle运行了3次才会运行此内容
24             @Override
25             public void run() {
26                 System.out.println("AfterTask;");
27             }
28         });
29     }
30 
31     public void test(){
32         runn();
33         for (int index = 0; index < 12; index ++){
34             excu.execute(new PreTask(index, barrier));
35         }
36     }
37 }
----------
test()运行结果：
AfterTask;
2,0,1,AfterTask;
3,4,5,AfterTask;
6,7,8,AfterTask;
11,9,10,
test()运行结果：

3、基础优先级队列PriorityBlockingQueue

可以根据优先级执行顺便执行，必须要实现Comparable接口；

 1 class PriorityTask implements Runnable, Comparable{
 2     private final int index;
 3     public final int priority = new Random().nextInt(20);
 4     PriorityTask(int index) {this.index = index;}
 5 
 6     @Override
 7     public void run() {
 8         System.out.println(index + " run with priority:" + priority);
 9     }
10     @Override
11     public int compareTo(Object o) {
12         return this.priority - ((PriorityTask) o).priority;
13     }
14 }
15 class Consumer implements Runnable{
16     private PriorityBlockingQueue<Runnable> queue;
17     Consumer(PriorityBlockingQueue<Runnable> queue) { this.queue = queue;}
18 
19     @Override
20     public void run() {
21         while (!Thread.interrupted()){
22             try {
23                 queue.take().run();
24             } catch (InterruptedException e) {}
25         }
26     }
27 }
28 
29 void test(){
30     ExecutorService excu = Executors.newCachedThreadPool();
31     PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<Runnable>();
32     for (int index = 1; index < 7; index ++){
33         queue.put(new PriorityTask(index));
34     }
35     excu.execute(new Consumer(queue));
36 }
----------
test()运行结果：
1 run with priority:1
6 run with priority:3
4 run with priority:8
2 run with priority:10
3 run with priority:15
5 run with priority:16
test()运行结果：

4、等待执行DelayQueue

等待一段时间后执行，可作超时处理任务之类的工作，其实除去等待时间也是一个优先级队列；

 1 class DelayTask implements Runnable, Delayed{
 2     private int index;
 3     int wait = new Random().nextInt(10);
 4     private Long startTime = System.currentTimeMillis();
 5     DelayTask(int index){ this.index = index;}
 6     public int getWait(){return this.wait;}
 7 
 8     //如果两个都在队列里判断谁优先执行(与getDelay的时间无关)，所以一般可以用延迟的时间进行排序
 9     @Override
10     public int compareTo(Delayed o) {
11         return wait - ((DelayTask)o).getWait();
12     }
13     //等待时间,一定要减去当前时间，因为此方法会被持续调用，直到返回值为0或负数时才执行run的方法，返回一个常量的正数，则run一直不会运行。
14     @Override
15     public long getDelay(TimeUnit unit) {
16         return unit.convert(startTime + wait * 1000  - System.currentTimeMillis(),
17                 TimeUnit.MILLISECONDS);
18     }
19     @Override
20     public void run() {
21         System.out.println(index + "waited:" + wait + "s,");
22     }
23 }
24 class Consumer implements Runnable{
25     private DelayQueue<DelayTask> queue;
26     Consumer(DelayQueue<DelayTask> queue) {this.queue = queue;}
27     @Override
28     public void run() {
29         while (!Thread.interrupted()){
30             try {
31                 queue.take().run();
32             } catch (InterruptedException e) {
33                 e.printStackTrace();
34             }
35         }
36     }
37 }
38 
39 void test(){
40     ExecutorService excu = Executors.newCachedThreadPool();
41     DelayQueue<DelayTask> queue = new DelayQueue<DelayTask>();
42     for (int index = 1; index < 12; index ++){
43         queue.put(new DelayTask(index));
44     }
45     excu.execute(new Consumer(queue));
46 }
----------
test()运行结果：
4waited:0s,
10waited:1s,
7waited:1s,
11waited:3s,
1waited:3s,
2waited:4s,
8waited:5s,
3waited:7s,
6waited:7s,
5waited:8s,
9waited:8s,
test()运行结果：

5、定时任务ScheduledThreadPoolExecutor

 1 class ScheduleTask implements Runnable{
 2     private final String name;
 3     ScheduleTask(String name) {this.name = name; }
 4 
 5     @Override
 6     public void run() {
 7         System.out.println(new Date().toString() + " running " + name);
 8     }
 9 }
10 void test() throws InterruptedException {
11     ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(10);//线程池容量
12     //执行一次
13     scheduler.schedule(new ScheduleTask("singletask"), 2, TimeUnit.SECONDS);
14     //周期性执行多次
15     scheduler.scheduleAtFixedRate(new ScheduleTask("repeatetask"), 1, 4, TimeUnit.SECONDS);
16     TimeUnit.SECONDS.sleep(20);
17     scheduler.shutdownNow();
18 }
----------
test()运行结果：
Thu Feb 04 01:08:19 CST 2021 running repeatetask
Thu Feb 04 01:08:20 CST 2021 running singletask
Thu Feb 04 01:08:23 CST 2021 running repeatetask
Thu Feb 04 01:08:27 CST 2021 running repeatetask
Thu Feb 04 01:08:31 CST 2021 running repeatetask
Thu Feb 04 01:08:35 CST 2021 running repeatetask
test()运行结果：

六、其他

1、变量在各线程内时独立的，使用java.lang.ThreadLocal<E>这个类来辅助实现线程本地存储
2、使用Lock通常会比使用synchronized高效许多，但一般可以 synchronized关键字入手，只有在性能调优时替换为Lock对象；Atomic对象只有在非常简单的情况下才有用，通常不使用。
3、线程池的创建方式：
Executors.newFixedThreadPool：创建一个固定大小的线程池，可控制并发的线程数，超出的线程会在队列中等待；
Executors.newCachedThreadPool：创建一个可缓存的线程池，若线程数超过处理所需，缓存一段时间后会回收，若线程数不够，则新建线程；
Executors.newSingleThreadExecutor：创建单个线程数的线程池，它可以保证先进先出的执行顺序；
Executors.newScheduledThreadPool：创建一个可以执行延迟任务的线程池；
Executors.newSingleThreadScheduledExecutor：创建一个单线程的可以执行延迟任务的线程池；
Executors.newWorkStealingPool：创建一个抢占式执行的线程池（任务执行顺序不确定）【JDK 1.8 添加】。
ThreadPoolExecutor：最原始的创建线程池的方式，它包含了 7 个参数可供设置，后面会详细讲

参考：线程池的7种创建方式

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/aland-1415/p/14449745.html