为什么要使用多线程

应用场景

  • 秒杀

    1.  思考：图书秒杀案例，为什么在高并发下会出问题？<br />项目代码：[https://gitee.com/eleorc/seckill.git](https://gitee.com/eleorc/seckill.git)<br />Apache压力测试工具：[https://www.apachehaus.com/cgi-bin/download.plx](https://www.apachehaus.com/cgi-bin/download.plx)

  • 抢票

经典抢票案例。

  • 同步代码的粒度
  • 双重检查 ```java package com.qf.sy2103.thread02;

import java.util.concurrent.TimeUnit;

/**

  • 并发问题的根源：
  • 如果程序中
  • （1）存在共享资源（实例对象的实例属性），
  • （2）并且多个线程可以同时访问到该共享资源，
  • （3）并且访问共享资源的方法不能保证原子性
  • 就会出现并发问题。

  • 解决方案：采用适当的同步措施。 */

public class SaleTicket { public static void main(String[] args) {

  1.     Ticket ticket = new Ticket();
  2.     int i = 0;
  3.     for (; i < 100; i++) {
  4.         new Thread(new Runnable() {
  5.             @Override
  6.             public void run() {
  7.                 try {
  8.                     TimeUnit.SECONDS.sleep(1);
  9.                 } catch (InterruptedException e) {
  10.                     e.printStackTrace();
  11.                 }
  12.                 for (int i = 0; i < 30; i++) {
  13.                     ticket.sale();
  14.                 }
  15.             }
  16.         }).start();
  17.     }
  18. }

}

class Ticket {

  1. private int number = 5;// 初始有5张票
  3. public void sale() {
  4.     if (number > 0) {
  5.         synchronized (this) {
  6.             if (number > 0) {
  7.                 System.out.println(Thread.currentThread().getName() + ":出票，票号" + (number--));
  8.             }
  9.         }
  10.     }
  11. }

}

  2.    - 一个奇怪的结果，同一个线程似乎有较大的概率获取到锁（非公平锁）
  3.    - synchronized是非公平锁
  4.    - 公平锁？
  5. ```java
  6. package com.qf.sy2103.seckill;
  8. import java.util.concurrent.TimeUnit;
  10. public class SaleTicket {
  11.     public static void main(String[] args) {
  13.         Ticket ticket = new Ticket();
  14.         int i = 0;
  15.         for (;i<100;i++) {
  16.             new Thread(new Runnable() {
  17.                 @Override
  18.                 public void run() {
  19.                     try {
  20.                         TimeUnit.SECONDS.sleep(1);
  21.                     } catch (InterruptedException e) {
  22.                         e.printStackTrace();
  23.                     }
  24.                     for (int i = 0;i<30;i++) {
  25.                         ticket.sale();
  26.                     }
  27.                 }
  28.             }).start();
  29.         }
  30.     }
  31. }
  33. class Ticket{
  35.     private int number = 5 ;// 初始有5张票
  37.     public void sale(){
  38.         if(number>0){
  39.             System.out.println(Thread.currentThread().getName()+":出票，票号"+(number--));
  40.         }
  41.     }
  43. }
  • 大数据处理
  • 生产者消费者模型：使用多线程可以简化程序开发，典型的就是生产者消费者模型。生产者和消费者使用不同的线程。

多线程理想的状态：无冲突并行

多线程编程模型的引入

web编程中自动引入了多线程模型

JavaAPI中自动引入多线程模型

如何控制多线程程序的正确性？加锁

利用JUC包的类自定义锁

  1. package com.qf.sy2103.seckill;
  3. import java.util.concurrent.ConcurrentLinkedQueue;
  4. import java.util.concurrent.atomic.AtomicInteger;
  5. import java.util.concurrent.locks.LockSupport;
  7. public class QfLock {
  9.     private AtomicInteger state = new AtomicInteger(0);
  11.     private Thread owner ;
  13.     private ConcurrentLinkedQueue<Thread> queue = new ConcurrentLinkedQueue<>();
  15.     public void lock(){
  16.         if (tryAquire()){
  17.             return;
  18.         }
  19.         Thread currentThread = Thread.currentThread();
  20.         for (;;){
  21.             if (tryAquire()){
  22.                 return;
  23.             }
  24.             queue.add(currentThread);
  25.             LockSupport.park(currentThread);
  26.         }
  27.     }
  29.     private boolean tryAquire() {
  30.         final boolean b = state.compareAndSet(0, 1);
  31.         if (b){
  32.             this.owner = Thread.currentThread();
  33.             return true;
  34.         }else {
  35.             return false;
  36.         }
  37.     }
  39.     public void unlock(){
  40.         if (this.owner!=Thread.currentThread()){
  41.             throw new RuntimeException("!!!");
  42.         }
  43.         final boolean b = state.compareAndSet(1, 0);
  44.         if (b){
  45.             this.owner = null;
  46.             final Thread next = queue.poll();
  47.             if (next!=null){
  48.                 LockSupport.unpark(next);
  49.             }
  50.         }
  51.     }
  54.     public static void main(String[] args) {
  55.         final QfLock qfLock = new QfLock();
  57.         for (int i=0;i<3;i++) {
  58.             new Thread(new Runnable() {
  59.                 @Override
  60.                 public void run() {
  61.                     qfLock.lock();
  62.                     System.out.println(Thread.currentThread().getName());
  63.                     qfLock.unlock();
  64.                 }
  65.             }).start();
  66.         }
  67.     }
  69. }

公平锁、非公平锁

  1. 公平锁：当有多个线程进行申请同一把锁的时候，如果线程是按照申请锁的先后顺序依次获取到该锁的，这种锁就叫做公平锁。a,b,c ,假如a线程拿到了synchronized锁，d线程进入争抢锁了，公平锁会保证，一定是b接下来拿到锁，然后是c接下来拿到锁，然后才是d。
  2. 非公平锁：synchronized 。 a,b,c ,加入a线程拿到了synchronized锁，d线程进入争抢锁了，d就竞争到了锁，那么b和c依然在等待，d可以运行了。 ```java private Lock lock = new ReentrantLock();// 创建一个juc包里的可重入锁 非公平锁

// private Lock lock = new ReentrantLock(true); // 创建一个公平锁。

  1. <a name="TgXP2"></a>
  2. ### 可重入锁
  3. ```java
  4. package com.qf.sy2103.thread02;
  6. /**
  7.  * synchronized 是可重入锁 
  8.  */
  9. public class ReentrantLockDemo {
  11.     public static void main(String[] args) {
  12.         final ReentrantLockDemo reentrantLockDemo = new ReentrantLockDemo();
  13.         reentrantLockDemo.test3();
  14.     }
  17.     public synchronized void test1(){
  18.         System.out.println("test1 is runnning..");
  19.         test2();
  20.     }
  22.     public synchronized void test2(){
  23.         System.out.println("test2 is runnning..");
  24.     }
  26.     public void test3(){
  27.         test1();
  28.     }
  30. }
  1. package com.qf.sy2103.thread02;
  3. import java.util.concurrent.locks.ReentrantLock;
  5. /**
  6.  * ReentrantLock 是可重入锁
  7.  */
  8. public class ReentrantLockDemo {
  10.     private ReentrantLock lock = new ReentrantLock();
  13.     public static void main(String[] args) {
  14.         final ReentrantLockDemo reentrantLockDemo = new ReentrantLockDemo();
  15.         reentrantLockDemo.test3();
  16.     }
  19.     public  void test1(){
  20.         System.out.println("test1 is runnning..");
  21.         test2();
  22.     }
  24.     public  void test2(){
  25.         System.out.println("test2 is runnning..");
  26.     }
  28.     public void test3(){
  29.         lock.lock();
  30.         lock.lock();
  31.         try {
  32.             test1();
  33.         } catch (Exception e) {
  34.             e.printStackTrace();
  35.         } finally {
  36.             lock.unlock();
  37.             lock.unlock();
  38.         }
  39.     }
  41. }

自旋锁

自旋：一段死循环 。

  1. package com.qf.juc;
  3. import java.util.concurrent.atomic.AtomicReference;
  5. /**
  6.  * 自定义自旋锁
  7.  */
  8. public class MySpinLock {
  10.     private AtomicReference<Thread> atomicReference = new AtomicReference<>();
  12.     public void lock(){
  14.         while ( !atomicReference.compareAndSet(null,Thread.currentThread()) ){
  15.             System.out.println("....");
  16.         }
  17.     }
  19.     public void unlock(){
  20.         while ( !atomicReference.compareAndSet(Thread.currentThread(),null) ){
  22.         }
  23.     }
  25.     public static void main(String[] args) {
  26.         MySpinLock mySpinLock = new MySpinLock();
  27.         mySpinLock.lock();
  28.         System.out.println("tets");
  29.         mySpinLock.unlock();
  30.     }
  32. }

读写锁

  1. package com.qf.sy2103.thread02;
  3. import java.util.HashMap;
  4. import java.util.concurrent.TimeUnit;
  5. import java.util.concurrent.locks.ReentrantReadWriteLock;
  7. public class ReadWriteLockDemo {
  9.     private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  11.     private ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
  13.     private ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
  15.     private HashMap<Integer,Integer> map = new HashMap<Integer,Integer>();
  17.     public void put(Integer key,Integer value){
  18.         try {
  19.             writeLock.lock();
  20.             System.out.println(Thread.currentThread().getName()+"write ...");
  21.             TimeUnit.SECONDS.sleep(3);
  22.             map.put(key,value);
  23.         } catch (Exception e) {
  24.             e.printStackTrace();
  25.         } finally {
  26.             writeLock.unlock();
  27.         }
  28.     }
  30.     public Integer get(Integer key){
  31.         Integer v = null;
  32.         try {
  33.             readLock.lock();
  34.             System.out.println(Thread.currentThread().getName()+"read...");
  35.             TimeUnit.SECONDS.sleep(1);
  36.             v =  map.get(key);
  37.         } catch (Exception e) {
  38.             e.printStackTrace();
  39.         } finally {
  40.             readLock.unlock();
  41.         }
  42.         return v;
  43.     }
  46.     public static void main(String[] args) {
  48.         final ReadWriteLockDemo readWriteLockDemo = new ReadWriteLockDemo();
  50.         new Thread(new Runnable() {
  51.             @Override
  52.             public void run() {
  53.                 for (int i = 0; i <3 ; i++) {
  54.                     readWriteLockDemo.put(i,i);
  55.                 }
  56.             }
  57.         }).start();
  60.         for (int i = 0 ;i<5;i++) {
  61.             new Thread(new Runnable() {
  62.                 @Override
  63.                 public void run() {
  64.                     for (int i = 0 ;i<3;i++) {
  65.                         final Integer integer = readWriteLockDemo.get(1);
  66.                     }
  67.                 }
  68.             }).start();
  69.         }
  71.         new Thread(new Runnable() {
  72.             @Override
  73.             public void run() {
  74.                 for (int i = 0; i <3 ; i++) {
  75.                     readWriteLockDemo.put(i,i);
  76.                 }
  77.             }
  78.         }).start();
  81.     }
  83. }

死锁

检测死锁

  1. jps : 列出本机所有的java进程号
  2. jstack 进程号： 列出死锁信息 ```java package com.qf.juc;

import java.util.concurrent.TimeUnit;

public class DeadLock {

  1. public static void main(String[] args) {
  3.     Object locka = new Object();
  4.     Object lockb = new Object();
  6.     new Thread(new Runnable() {
  7.         @Override
  8.         public void run() {
  9.             synchronized (locka) {
  10.                 System.out.println(Thread.currentThread().getName() + "获取到了locka");
  11.                 try {
  12.                     TimeUnit.SECONDS.sleep(1);
  13.                     synchronized (lockb) {
  14.                         System.out.println(Thread.currentThread().getName() + "获取到了lockb");
  15.                     }
  16.                 } catch (InterruptedException e) {
  17.                     e.printStackTrace();
  18.                 }
  19.             }
  20.         }
  21.     }, "A").start();
  24.     new Thread(new Runnable() {
  25.         @Override
  26.         public void run() {
  27.             synchronized (lockb) {
  28.                 System.out.println(Thread.currentThread().getName() + "获取到了lockb");
  30.                 synchronized (locka) {
  31.                     System.out.println(Thread.currentThread().getName() + "获取到了locka");
  32.                 }
  33.             }
  34.         }
  35.     }, "B").start();
  37. }

}

  1. <a name="AIU7m"></a>
  2. ## 多线程协作编程模型
  3. <a name="w7p3h"></a>
  4. ### 生产者和消费者问题
  5. 口诀： 判断、等待、业务、通知
  6. ```java
  7. package com.qf.juc;
  9. public class ProduceConsum {
  11.     public static void main(String[] args) {
  12.         Product product = new Product();
  14.         new Thread(new Runnable() {
  15.             @Override
  16.             public void run() {
  17.                 try {
  18.                     for (int i=0;i<10;i++) {
  19.                         product.add();
  20.                     }
  21.                 } catch (InterruptedException e) {
  22.                     e.printStackTrace();
  23.                 }
  24.             }
  25.         },"A").start();
  27.         new Thread(new Runnable() {
  28.             @Override
  29.             public void run() {
  30.                 try {
  31.                     for (int i=0;i<10;i++) {
  32.                         product.sub();
  33.                     }
  34.                 } catch (InterruptedException e) {
  35.                     e.printStackTrace();
  36.                 }
  37.             }
  38.         },"B").start();
  41.     }
  43. }
  45. class Product {
  47.     private int count = 0;
  49.     // 生产商品，如果商品的数量大于0，则停止生产商品
  50.     public synchronized void add() throws InterruptedException {
  51.         // 1 判断
  52.         if(count > 0) {
  53.             this.wait();  // 2 等待
  54.         }
  55.         // 3 、业务代码处理
  56.         System.out.println(Thread.currentThread().getName()+":生产了一个商品，当前的商品数量："+(++count));
  57.         // 4 、通知其他等待线程
  58.         this.notifyAll();
  59.     }
  61.     // 消费商品，如果商品的数量小于1，则停止消费商品
  62.     public synchronized void sub() throws InterruptedException {
  63.         // 1 判断
  64.         if(count<1){
  65.             this.wait(); // 2 等待
  66.         }
  67.         // 3 业务
  68.         System.out.println(Thread.currentThread().getName()+":消费了一个商品，当前的商品数量为："+(--count));
  69.         // 4 通知其他线程
  70.         this.notifyAll();
  71.     }
  73. }

虚假唤醒问题

上面的示例如果生产者和消费者的线程不是一比一，则会出现问题。

  1. class Product {
  3.     private int count = 0;
  5.     // 生产商品，如果商品的数量大于0，则停止生产商品
  6.     public synchronized void add() throws InterruptedException {
  7.         // 1 判断
  8.         while (count > 0) {
  9.             this.wait();  // 2 等待
  10.         }
  11.         // 3 、业务代码处理
  12.         System.out.println(Thread.currentThread().getName()+":生产了一个商品，当前的商品数量："+(++count));
  13.         // 4 、通知其他等待线程
  14.         this.notifyAll();
  15.     }
  17.     // 消费商品，如果商品的数量小于1，则停止消费商品
  18.     public synchronized void sub() throws InterruptedException {
  19.         // 1 判断
  20.         while(count<1){
  21.             this.wait(); // 2 等待
  22.         }
  23.         // 3 业务
  24.         System.out.println(Thread.currentThread().getName()+":消费了一个商品，当前的商品数量为："+(--count));
  25.         // 4 通知其他线程
  26.         this.notifyAll();
  27.     }
  29. }

精准通知问题

  1. package com.qf.juc;
  3. import java.util.concurrent.locks.Condition;
  4. import java.util.concurrent.locks.Lock;
  5. import java.util.concurrent.locks.ReentrantLock;
  7. public class Demo5 {
  9.     public static void main(String[] args) {
  10.         A a = new A();
  11.         new Thread(new Runnable() {
  12.             @Override
  13.             public void run() {
  14.                 for (int i =0;i<5;i++) {
  15.                     a.test1();
  16.                 }
  17.             }
  18.         },"A").start();
  20.         new Thread(new Runnable() {
  21.             @Override
  22.             public void run() {
  23.                 for (int i=0;i<5;i++) {
  24.                     a.test2();
  25.                 }
  26.             }
  27.         },"B").start();
  29.         new Thread(new Runnable() {
  30.             @Override
  31.             public void run() {
  32.                 for (int i=0;i<5;i++) {
  33.                     a.test3();
  34.                 }
  35.             }
  36.         },"C").start();
  37.     }
  39. }
  42. class A {
  44.     private int state = 1;
  46.     private Lock lock = new ReentrantLock();
  47.     private Condition condition1 = lock.newCondition();
  48.     private Condition condition2 = lock.newCondition();
  49.     private Condition condition3 = lock.newCondition();
  51.     public void test1() {
  52.         // 1 加锁
  53.         lock.lock();
  54.         try {
  55.             while (state != 1) {
  56.                 // 让当前线程在 condition1 上进行等待
  57.                 condition1.await();
  58.             }
  59.             System.out.println(Thread.currentThread().getName() + "test1 start... ");
  60.             // 指定唤醒的test2 这个方法
  61.             this.state = 2;
  62.             // 定向通知，通知在condition2 上等待的线程，可以唤醒执行了
  63.             condition2.signal();
  65.         } catch (InterruptedException e) {
  66.             e.printStackTrace();
  67.         } finally {
  68.             lock.unlock();
  69.         }
  71.     }
  73.     public void test2() {
  74.         // 1 加锁
  75.         lock.lock();
  76.         try {
  77.             while (state != 2) {
  78.                 // 让当前线程在 condition1 上进行等待
  79.                 condition2.await();
  80.             }
  81.             System.out.println(Thread.currentThread().getName() + "test2 start... ");
  82.             // 指定唤醒的test2 这个方法
  83.             this.state = 3;
  84.             // 定向通知，通知在condition3 上等待的线程，可以唤醒执行了
  85.             condition3.signal();
  87.         } catch (InterruptedException e) {
  88.             e.printStackTrace();
  89.         } finally {
  90.             lock.unlock();
  91.         }
  93.     }
  95.     public void test3() {
  96.         // 1 加锁
  97.         lock.lock();
  98.         try {
  99.             while (state != 3) {
  100.                 // 让当前线程在 condition1 上进行等待
  101.                 condition3.await();
  102.             }
  103.             System.out.println(Thread.currentThread().getName() + "test3 start... ");
  104.             // 指定唤醒的test1 这个方法
  105.             this.state = 1;
  106.             // 定向通知，通知在condition1 上等待的线程，可以唤醒执行了
  107.             condition1.signal();
  109.         } catch (InterruptedException e) {
  110.             e.printStackTrace();
  111.         } finally {
  112.             lock.unlock();
  113.         }
  114.     }
  115. }

JUC常用辅助类

CountDownLatch

  1. package com.qf.juc;
  3. import java.util.concurrent.CountDownLatch;
  4. import java.util.concurrent.TimeUnit;
  6. public class Demo2 {
  7.     // 共享资源
  8.     public int a = 0 ;
  10.     public static void main(String[] args) throws InterruptedException {
  12.         Demo2 demo2 = new Demo2();
  14.         // 创建一个倒计时器，并且赋值为2.如果主线程在latch对象上等待的化，那么需要其他线程
  15.         // 调用2次  countdown方法，主线程才会被唤醒继续执行。
  16.         CountDownLatch latch = new CountDownLatch(2);
  18.         new Thread(new Runnable() {
  19.             @Override
  20.             public void run() {
  21.                 for (int i=0;i<100000;i++) {
  22.                     demo2.a++;
  23.                 }
  24.                 // 把latch对象的计数器减1
  25.                 latch.countDown();
  26.             }
  27.         },"A").start();
  29.         new Thread(new Runnable() {
  30.             @Override
  31.             public void run() {
  32.                 for (int i=0;i<100000;i++) {
  33.                     demo2.a++;
  34.                 }
  35.                 latch.countDown();
  36.             }
  37.         },"B").start();
  39.         // 让main线程等待一会
  40. //        Thread.sleep(1000);
  41.         // todo: juc 里有单独的类解决这个问题 ： CountDownLatch
  42. //        TimeUnit.SECONDS.sleep(10);
  43.         // 让主线程在 latch对象上等待，那么需要等待多久呢？
  44.         // 需要等到 latch对象的计数器变成0 .
  45.         latch.await();
  47.         System.out.println(demo2.a);
  49.     }
  51. }

CyclicBarrier （召唤神龙）

  1. package com.qf.juc;
  3. import java.util.concurrent.BrokenBarrierException;
  4. import java.util.concurrent.CyclicBarrier;
  6. public class Demo7 {
  8.     public static void main(String[] args) {
  10.         CyclicBarrier cyclicBarrier = new CyclicBarrier(7, new Runnable() {
  11.             @Override
  12.             public void run() {
  13.                 System.out.println("集齐了七颗龙珠可以召唤神龙了！");
  14.             }
  15.         });
  18.         for (int i=0;i<7;i++) {
  19.             int finalI = i;
  20.             new Thread(new Runnable() {
  21.                 @Override
  22.                 public void run() {
  23.                     System.out.println(Thread.currentThread().getName()+":收集到了龙珠："+ finalI);
  24.                     try {
  25.                         cyclicBarrier.await();
  26.                         System.out.println(Thread.currentThread().getName()+":执行结束");
  27.                     } catch (InterruptedException e) {
  28.                         e.printStackTrace();
  29.                     } catch (BrokenBarrierException e) {
  30.                         e.printStackTrace();
  31.                     }
  32.                 }
  33.             },""+i).start();
  34.         }
  36.     }
  38. }

Semaphore

  1. package com.qf.juc;
  3. import java.util.concurrent.Semaphore;
  4. import java.util.concurrent.TimeUnit;
  6. public class Demo8 {
  8.     public static void main(String[] args) {
  10.         Semaphore semaphore = new Semaphore(3);
  12.         for (int i=0;i<6;i++) {
  13.             new Thread(new Runnable() {
  14.                 @Override
  15.                 public void run() {
  16.                     for (int i=0;i<5;i++) {
  17.                         try {
  18.                             // 在信号量上申请，如果成功，可以执行业务代码，并且信号两的计数器会减一
  19.                             semaphore.acquire();
  20.                             System.out.println(Thread.currentThread().getName()+":抢到了车位");
  21.                             TimeUnit.SECONDS.sleep(1);
  22.                         } catch (InterruptedException e) {
  23.                             e.printStackTrace();
  24.                         }finally {
  25.                             // 信号两的计数器加1
  26.                             semaphore.release();
  27.                         }
  28.                     }
  30.                 }
  31.             },""+i).start();
  32.         }
  35.     }
  37. }