juc多线程编程学习

Posted 2020-11-22 tdyang

JUC是java.util.concurrent的缩写，java.util.concurrent是在并发编程中使用的工具类。

 

 

 

 

在以前的解决并发问题，一般是通过Synchronize关键字，现在可以通过juc下的工具类，来解决多线程并发问题。

 

 

首先写有个demo：使用synchronized进行上锁

 

 

public class Synchronizedemo {

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

    }
}
class Ticket{

    private int num = 30;

    public  synchronized void sale(){
        if(num > 0){
            System.out.println("线程"+Thread.currentThread().getName()+",卖出第"+num+"票");
        }
        num--;
    }
}

Lock:Lock是一个接口

这次我们先使用lock锁,实现提供了比使用同步方法和语句可以获得的更广泛的锁操作。

它们允许更灵活的结构，可能具有非常不同的属性，并且可能支持多个关联的条件对象。

根据api手册：我们使用ReentrantLock这个实现类来操作锁，这个锁被称为可重入锁，可以理解为厕所门栓。

 

It is recommended practice to always immediately follow a call to lock with a try block, most typically in a before/after construction such as:

 
 class X {
   private final ReentrantLock lock = new ReentrantLock();
   // ...

   public void m() {
     lock.lock();  // block until condition holds
     try {
       // ... method body
     } finally {
       lock.unlock()
     }
   }
 }

 

现在用它来写卖票程序:最后我们需要释放锁资源。

package com.study.lock;


import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

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

class Ticket{

    private int num = 30;
    Lock lock = new ReentrantLock();

    public  void sale(){

        lock.lock();
        try{
            if(num > 0){
                System.out.println(Thread.currentThread().getName()+"卖出了第"+num--+"还剩有"+(num)+"张票");
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

 

再来另外一个例子：实现一个线程对该变量加1，一个线程对该变量减1，循环交替。

demo1：使用Synchronized实现

class Test{
    private int num = 0 ;

    public synchronized void increase() throws InterruptedException {

       if (num != 0){   
            this.wait();
        }
        num++;
        System.out.println(Thread.currentThread().getName()+","+num);
        this.notifyAll();
    }
    public synchronized void  desc() throws InterruptedException {

       if (num == 0){
            this.wait();
        }
        num--;
        System.out.println(Thread.currentThread().getName()+","+num);
        this.notifyAll();
    }
}

public class ProduConsumer {

    public static void main(String[] args) throws Exception {

        Test test = new Test();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.increase();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"a").start();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.desc();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"b").start();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.increase();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"c").start();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.desc();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"d").start();
    }
}

根据打印结果，我们知道这样会导致数据发生错误，那么这是为什么呢？根据官方文档我们可知，打断和虚假唤醒是可能的，所以我们应该使用循环，而不是if

As in the one argument version, interrupts and spurious wakeups are possible, and this method should always be used in a loop:

     synchronized (obj) {
         while (<condition does not hold>)
             obj.wait(timeout, nanos);
         ... // Perform action appropriate to condition
     }
 

所以将while替换成if，为什么使用if不行？想一下，如果我们的a线程，进入判断后，执行到 this.wait()方法中断 ; 这时候，c线程一样。进入判断，也中断。

这时候就有两个线程都在等待wait，然后b线程notifyAll对a,c线程唤醒，但由于判断条件是if,他们没有回头再去判断，这时候两个线程都执行num++,所以会出现2了，

所以需要使用while,让它重新回来判断，防止虚假唤醒。

 while (num != 0){
           this.wait();
       }
num++;

demo2：使用lock实现

package com.study.prodConsumer;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class Testdemo{
    private int num = 0 ;
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    public  void increase() throws InterruptedException {

         lock.lock();
         try{
             while (num != 0){
                 condition.await();
             }
             num++;
             System.out.println(Thread.currentThread().getName()+","+num);
             condition.signalAll();
         }catch (Exception e){
             e.printStackTrace();
         }finally {
             lock.unlock();
         }
    }

    public  void  desc() throws InterruptedException {

        lock.lock();
        try{
            while (num != 0){
                condition.await();
            }
            num--;
            System.out.println(Thread.currentThread().getName()+","+num);
            condition.signalAll();

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }

    }

}

public class ProduConsumerLock {
    public static void main(String[] args) {
        Test test = new Test();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.increase();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"a").start();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.desc();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"b").start();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.increase();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"c").start();

        new Thread(()->{
            for (int i = 0 ; i < 10 ; i++){
                try {
                    test.desc();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"d").start();
    }
}

Synchronized的wait和lock的await对应，notifyAll和signalAll对应,这里我们需要使用Condition。根据官网对lock的解释：

锁实现提供了比使用同步方法和语句更广泛的锁操作。它们允许更灵活的结构，可能具有完全不同的属性，并且可能支持
多个关联的条件对象。

 

 

Lock implementations provide more extensive locking operations than can be obtained using synchronized methods and statements. 
They allow more flexible structuring, may have quite different properties, and may support multiple associated Condition objects.

 

所以我们看看Condition:大概意思就是可以使用Condition来代替Object monitor methods,也就是wait, notify and notifyAll方法，相当于Condition就是Lock的钥匙。

 

Condition factors out the Object monitor methods (wait, notify and notifyAll) into distinct objects to give the effect of having multiple wait-sets per object, 
by combining them with the use of arbitrary Lock implementations. 
Where a Lock replaces the use of synchronized methods and statements, a Condition replaces the use of the Object monitor methods.

官方写了一个实例告诉我们怎么使用：

 class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition(); 
   final Condition notEmpty = lock.newCondition(); 

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock();
     try {
       while (count == items.length)
         notFull.await();
       items[putptr] = x;
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal();
     } finally {
       lock.unlock();
     }
   }

   public Object take() throws InterruptedException {
     lock.lock();
     try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally {
       lock.unlock();
     }
   }
 }

 

lock锁还有一大好处就是可以精确通知唤醒：

demo：多线程之间按顺序调用，实现A->B->C 三个线程启动，A打印5次，B打印10次，C打印15次

class  ConditionTest{
    private int num = 1;//标识

    private Lock lock = new ReentrantLock();
    //一把锁配三把钥匙
    private Condition condition1 = lock.newCondition();
    private Condition condition2 = lock.newCondition();
    private Condition condition3 = lock.newCondition();

    public void print5(){
         lock.lock();
         try{
            while (num != 1){
                condition1.await();
            }
             for (int i = 0; i < 5 ; i++) {
                 System.out.println(Thread.currentThread().getName()+(i+1));
             }
            num = 2;
            condition2.signal();
         }catch (Exception e){
             e.printStackTrace();
         }finally {
             lock.unlock();
         }
    }

    public void print10(){
        lock.lock();
        try{
            while (num != 2){
                condition2.await();
            }
            for (int i = 0; i < 10 ; i++) {
                System.out.println(Thread.currentThread().getName()+(i+1));
            }
            num = 3;
            condition3.signal();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    public void print15(){
        lock.lock();
        try{
            while (num != 3){
                condition3.await();
            }
            for (int i = 0; i < 15 ; i++) {
                System.out.println(Thread.currentThread().getName()+(i+1));
            }
            num = 1;
            condition1.signal();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

public class ConditionDemo {
    public static void main(String[] args) {
        ConditionTest conditionTest = new ConditionTest();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                conditionTest.print5();
            }
        },"A").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                conditionTest.print10();
            }
        },"B").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                conditionTest.print15();
            }
        },"C").start();

    }
}

Synchronized和lock两者区别：
1.首先synchronized是java内置关键字，在jvm层面，Lock的实现类是个java类；
2.synchronized无法判断是否获取锁的状态，Lock可以判断是否获取到锁；
3.synchronized会自动释放锁(a 线程执行完同步代码会释放锁 ；b 线程执行过程中发生异常会释放锁)，Lock需在finally中手工释放锁（unlock()方法释放锁），否则容易造成线程死锁；
4.用synchronized关键字的两个线程1和线程2，如果当前线程1获得锁，线程2线程等待。如果线程1阻塞，线程2则会一直等待下去，而Lock锁就不一定会等待下去，如果尝试获取不到锁，线程可以不用一直等待就结束了；
5.synchronized的锁可重入、不可中断、非公平，而Lock锁可重入、可判断、可公平（两者皆可）
6.Lock锁适合大量同步的代码的同步问题，synchronized锁适合代码少量的同步问题。

 

在学习集合时，我们知道ArrayList和Hashset是线程不安全的，里面有很多操作公共属性的方法，而且方法上都没有进行加锁操作，所以当多线程并发的时候，会造成数据的错误。

那么怎么解决线称安全？看demo：

public class ListTest {

    public static void main(String[] args) {

       List<String> list1= Collections.synchronizedList(new ArrayList<>()); //Collections集合工具类 线程安全
         List<String> list = new Vector<>();//线程安全
       

        for(int i = 0 ; i < 30 ; i++){

            new Thread(()->{
                list.add(UUID.randomUUID().toString().substring(0,8));
                System.out.println(list);
            },"thread"+i).start();
        }
    }
}

我们可以使用

1》Collections集合工具类对List进行加锁，Collections.synchronizedList

2》可以使用线程安全类Vector

3》CopyOnWriteArrayList

CopyOnWriteArrayList是juc下提供了一种线程安全的集合，称为写时复制，它里面的add方法使用了可重入锁；remove方法也用了可重入锁

 原理：CopyOnWrite容器即写时复制的容器。往一个容器添加元素的时候，不直接往当前容器Object[]添加，
而是先将当前容器Object[]进行Copy，复制出一个新的容器Object[] newElements，然后向新的容器Object[] newElements里添加元素。
添加元素后，再将原容器的引用指向新的容器setArray(newElements)。
这样做的好处是可以对CopyOnWrite容器进行并发的读，而不需要加锁，因为当前容器不会添加任何元素。
所以CopyOnWrite容器也是一种读写分离的思想，读和写不同的容器。
 

public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            newElements[len] = e;
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }

 同理hashSet也有这样一个线程安全的CopyOnWriteArraySet

Set<String> set = new CopyOnWriteArraySet<>();//线程安全
 

HashMap是不安全的，juc也提供了ConcurrentHashMap保证线程安全

Map<String,String> map = new ConcurrentHashMap<>();//线程安全
 

 FurureTask/Callable：java中创建线程的方式

 

JUC中还提供了其他的功能强大的辅助类：

CountDownLatch：倒计时计数器

看demo:倒计时计数

class Test{
    public void start() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for (int i = 5; i >= 1; i--) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName());
                countDownLatch.countDown();
            },""+i).start();
        }
        countDownLatch.await();
        System.out.println("预备跑");
    }
}
public class countDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        Test test = new Test();
        test.start();
    }
}

原理：

 * CountDownLatch主要有两个方法，当一个或多个线程调用await方法时，这些线程会阻塞。
 * 其它线程调用countDown方法会将计数器减1(调用countDown方法的线程不会阻塞)，
 * 当计数器的值变为0时，因await方法阻塞的线程会被唤醒，继续执行。

CyclicBarrier：循环栅栏或加法计数器

demo：七个葫芦娃合体

public class CycleBarrierDemo {

    public static void main(String[] args) {

        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{ System.out.println("**葫芦娃合体**"); });

        for (int i = 0; i < 7 ; i++) {
            final int tem = i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"第"+tem+"葫芦娃");
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

原理：

 * CyclicBarrier
 * 的字面意思是可循环（Cyclic）使用的屏障（Barrier）。它要做的事情是，
 * 让一组线程到达一个屏障（也可以叫同步点）时被阻塞，
 * 直到最后一个线程到达屏障时，屏障才会开门，所有
 * 被屏障拦截的线程才会继续干活。
 * 线程进入屏障通过CyclicBarrier的await()方法。

 Semaphore：信号灯

 demo：七辆车占三个车位

public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);//三个车位

        for (int i = 0; i < 7 ; i++) {
            new Thread(()->{
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"抢到了车位");
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName()+"离开了车位");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    semaphore.release();
                }
            },""+i).start();
        }
    }

原理：

 在信号量上我们定义两种操作：
 * acquire（获取） 当一个线程调用acquire操作时，它要么通过成功获取信号量（信号量减1），
 *       要么一直等下去，直到有线程释放信号量，或超时。
 * release（释放）实际上会将信号量的值加1，然后唤醒等待的线程。
 *
 * 信号量主要用于两个目的，一个是用于多个共享资源的互斥使用，另一个用于并发线程数的控制。