多线程系列三：Lock和Condition

有了synchronized为什么还要Lock？

因为Lock和synchronized比较有如下优点，这些特点是synchronized没有的

1、 尝试非阻塞地获取锁

2、 获取锁的过程可以被中断

3、 超时获取锁  多长时间拿不到锁就放弃

Lock的使用范式：

package com.lgs;

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

// 显示锁Lock的使用范式
public class LockTemplete {

    public static void main(String[] args) {
        // 创建一把锁 ReentrantLock是Lock的实现类,表示可重入的锁
        Lock lock = new ReentrantLock();
        // 获取锁
        lock.lock();
        try {
            // 业务逻辑处理
        }finally {
            // 释放锁
            lock.unlock();
        }
    }
}

这里解释一下什么叫可重入锁：已经获得锁的线程再次进入加了锁的代码依然可以获取锁，比如递归同步代码块就要求锁是可重入的

Lock的常用方法

Lock() 获取锁

tryLock尝试非阻塞地获取锁

lockInterruptibly：获取锁的过程可以被中断

tryLock(long time, TimeUnit unit) 超时获取锁

unlock()释放锁

锁的可重入

一个线程获得了锁进入了同步代码块遇到了锁仍然可以获得锁进入同步代码块

递归的时候发生锁的重入，没有锁的可重入，就会死锁

公平锁和非公平锁

公平锁，先对锁发出获取请求的一定先被满足。公平锁的效率比非公平锁效率要低。

为什么非公平锁的性能要高：因为非公平锁是可以插队的，如线程C被唤醒变为可执行去获取锁的过程中，线程A插队进来直接获取锁执行自己的业务，线程A执行完以后，线程C刚好唤醒完直接就获取锁运行了，这样在线程C唤醒的过程中线程A就执行完了效率更高

读写锁ReentrantReadWriteLock

允许多个读线程同时进行，但是只允许一个写线程(不允许其他读线程是为了防止脏读)，支持读多写少场景，性能会有提升。

ReentrantReadWriteLock的使用范式：

package com.lgs;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

// 读写锁使用范式
public class ReadWriteLockTemplete {

    // 对map进行读写操作
    static final Map<String,String> map = new HashMap<>();
    
    // 定义一个读写锁
    static ReentrantReadWriteLock rwl = new  ReentrantReadWriteLock();
    
    // 拿到读锁
    static Lock r = rwl.readLock();
    
    // 拿到写锁
    static Lock w = rwl.writeLock();
    
    // 进行写操作
    public void put() {
        // 获取写锁
        w.lock();
        try {
            // 业务逻辑处理
        }finally {
            // 释放写锁
            w.unlock();
        }
    }
    
    // 进行写操作
    public void read() {
        // 获取读锁
        r.lock();
        try {
            // 业务逻辑处理
        }finally {
            // 释放读锁
            r.unlock();
        }
    }
}

ReentrantReadWriteLock和synchronized性能比较

案例：两种锁实现对库存的读写性能比较

定义一个货物库存的实体类:

package com.lgs.readwritelock.and.sync.perform.compare.model;


// 定义一个货物库存的实体类
public class GoodsVo {

    private final String id;
    private int totalSaleNumber;//总销售数
    private int depotNumber;//当前库存数

    public GoodsVo(String id, int totalSaleNumber, int depotNumber) {
        this.id = id;
        this.totalSaleNumber = totalSaleNumber;
        this.depotNumber = depotNumber;
    }

    public int getTotalSaleNumber() {
        return totalSaleNumber;
    }

    public int getDepotNumber() {
        return depotNumber;
    }

    // 设置总销售数和当前库存数
    public void setGoodsVoNumber(int changeNumber){
        this.totalSaleNumber += changeNumber;
        this.depotNumber -= changeNumber;
    }
}

定义一个货物库存的接口:

package com.lgs.readwritelock.and.sync.perform.compare.dao;

import com.lgs.readwritelock.and.sync.perform.compare.model.GoodsVo;

// 定义一个货物库存的接口
public interface IGoodsNum {

    // 获取货物数
    public GoodsVo getGoodsNumber();
    // 设置货物数
    public void setGoodsNumber(int changeNumber);
}

synchronized同步方式实现:

package com.lgs.readwritelock.and.sync.perform.compare.impl;

import com.lgs.readwritelock.and.sync.perform.compare.dao.IGoodsNum;
import com.lgs.readwritelock.and.sync.perform.compare.model.GoodsVo;

//synchronized同步方式实现
public class NumSynImpl implements IGoodsNum {

    private GoodsVo goods;

    public NumSynImpl(GoodsVo goods) {
        this.goods = goods;
    }


    // sync方式阻塞获取货物数
    @Override
    public synchronized GoodsVo getGoodsNumber() {
        try {
            Thread.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return this.goods;
    }

    // sync方式阻塞设置销售货物数
    @Override
    public synchronized void setGoodsNumber(int changeNumber) {

        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        this.goods.setGoodsVoNumber(changeNumber);

    }
}

ReentrantReadWriteLock可重如读写锁方式实现,可以多个线程读一个线程写:

package com.lgs.readwritelock.and.sync.perform.compare.impl;

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

import com.lgs.readwritelock.and.sync.perform.compare.dao.IGoodsNum;
import com.lgs.readwritelock.and.sync.perform.compare.model.GoodsVo;

// ReentrantReadWriteLock可重如读写锁方式实现,可以多个线程读一个线程写 
// 总体性能比synchronized更好
public class RwNumImpl implements IGoodsNum {

    private GoodsVo goods;

    private final  ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final  Lock r = lock.readLock();
    private final  Lock w = lock.writeLock();

    public RwNumImpl(GoodsVo goods) {
        this.goods = goods;
    }


    @Override
    public GoodsVo getGoodsNumber() {
        r.lock();
        try{
            try {
                Thread.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return this.goods;
        }finally{
            r.unlock();
        }
    }

    @Override
    public void setGoodsNumber(int changeNumber) {
        w.lock();
        try{
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            this.goods.setGoodsVoNumber(changeNumber);
        }finally{
            w.unlock();
        }
    }
}

运行测试方法比较性能，3个线程写 30个线程读

package com.lgs.readwritelock.and.sync.perform.compare.test;

import java.util.Random;
import java.util.concurrent.CountDownLatch;

import com.lgs.readwritelock.and.sync.perform.compare.dao.IGoodsNum;
import com.lgs.readwritelock.and.sync.perform.compare.impl.NumSynImpl;
import com.lgs.readwritelock.and.sync.perform.compare.impl.RwNumImpl;
import com.lgs.readwritelock.and.sync.perform.compare.model.GoodsVo;

//ReentrantReadWriteLock和synchronized性能比较
// 3个线程写 30个线程读
// 结论：ReentrantReadWriteLock性能比synchronized快8倍
public class Test {

    static final int threadBaseCount = 3;
    static final int threadRatio = 10;
    static CountDownLatch countDownLatch= new CountDownLatch(1);

    //模拟实际的数据库读操作
    private static class ReadThread implements Runnable{

        private IGoodsNum goodsNum;

        public ReadThread(IGoodsNum goodsNum) {
            this.goodsNum = goodsNum;
        }

        @Override
        public void run() {
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long start = System.currentTimeMillis();
            for(int i=0;i<100;i++){
                goodsNum.getGoodsNumber();
            }
            long duration = System.currentTimeMillis()-start;
            System.out.println(Thread.currentThread().getName()+"读取库存数据耗时："+duration+"ms");

        }
    }


    //模拟实际的数据库写操作
    private static class WriteThread implements Runnable{

        private IGoodsNum goodsNum;

        public WriteThread(IGoodsNum goodsNum) {
            this.goodsNum = goodsNum;
        }

        @Override
        public void run() {
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long start = System.currentTimeMillis();
            Random r = new Random();
            for(int i=0;i<10;i++){
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                goodsNum.setGoodsNumber(r.nextInt(10));
            }
            long duration = System.currentTimeMillis()-start;
            System.out.println(Thread.currentThread().getName()+"写库存数据耗时："+duration+"ms");

        }
    }

    public static void main(String[] args) throws InterruptedException {
        GoodsVo goodsVo =
                new GoodsVo("goods001",100000,10000);
        // synchronized方式
        // IGoodsNum goodsNum = new NumSynImpl(goodsVo);
        
        // ReentrantReadWriteLock方式
         IGoodsNum goodsNum = new RwNumImpl(goodsVo);
        
        //30个线程读
        for(int i = 0;i<threadBaseCount*threadRatio;i++){
            Thread readT = new Thread(new ReadThread(goodsNum));
            readT.start();
        }
        // 3个线程写
        for(int i = 0;i<threadBaseCount;i++){
            Thread writeT = new Thread(new WriteThread(goodsNum));
            writeT.start();
        }
        countDownLatch.countDown();

    }

}

synchronized方式耗时：

Thread-20读取库存数据耗时：1130ms
Thread-23读取库存数据耗时：6468ms
Thread-25读取库存数据耗时：8076ms
Thread-7读取库存数据耗时：9432ms
Thread-26读取库存数据耗时：10527ms
Thread-28读取库存数据耗时：10885ms
Thread-8读取库存数据耗时：11359ms
Thread-29读取库存数据耗时：11519ms
Thread-21读取库存数据耗时：14422ms
Thread-13读取库存数据耗时：14626ms
Thread-5读取库存数据耗时：14994ms
Thread-6读取库存数据耗时：15136ms
Thread-4读取库存数据耗时：15326ms
Thread-14读取库存数据耗时：15496ms
Thread-18读取库存数据耗时：15598ms
Thread-22读取库存数据耗时：15715ms
Thread-24读取库存数据耗时：15757ms
Thread-27读取库存数据耗时：15776ms
Thread-31写库存数据耗时：15963ms
Thread-3读取库存数据耗时：16889ms
Thread-17读取库存数据耗时：17250ms
Thread-32写库存数据耗时：17375ms
Thread-9读取库存数据耗时：17843ms
Thread-2读取库存数据耗时：17944ms
Thread-15读取库存数据耗时：18117ms
Thread-1读取库存数据耗时：18284ms
Thread-11读取库存数据耗时：18418ms
Thread-10读取库存数据耗时：18452ms
Thread-30写库存数据耗时：18515ms
Thread-0读取库存数据耗时：18814ms
Thread-12读取库存数据耗时：18897ms
Thread-16读取库存数据耗时：18990ms
Thread-19读取库存数据耗时：18995ms

ReentrantReadWriteLock方式耗时：

Thread-30写库存数据耗时：1899ms
Thread-32写库存数据耗时：1968ms
Thread-31写库存数据耗时：2032ms
Thread-13读取库存数据耗时：2479ms
Thread-7读取库存数据耗时：2480ms
Thread-2读取库存数据耗时：2481ms
Thread-4读取库存数据耗时：2487ms
Thread-18读取库存数据耗时：2490ms
Thread-15读取库存数据耗时：2491ms
Thread-23读取库存数据耗时：2489ms
Thread-17读取库存数据耗时：2490ms
Thread-10读取库存数据耗时：2491ms
Thread-11读取库存数据耗时：2491ms
Thread-5读取库存数据耗时：2492ms
Thread-19读取库存数据耗时：2490ms
Thread-16读取库存数据耗时：2490ms
Thread-12读取库存数据耗时：2491ms
Thread-24读取库存数据耗时：2489ms
Thread-14读取库存数据耗时：2491ms
Thread-20读取库存数据耗时：2490ms
Thread-0读取库存数据耗时：2502ms
Thread-28读取库存数据耗时：2494ms
Thread-1读取库存数据耗时：2498ms
Thread-22读取库存数据耗时：2496ms
Thread-8读取库存数据耗时：2497ms
Thread-21读取库存数据耗时：2496ms
Thread-9读取库存数据耗时：2497ms
Thread-27读取库存数据耗时：2495ms
Thread-26读取库存数据耗时：2495ms
Thread-3读取库存数据耗时：2499ms
Thread-29读取库存数据耗时：2494ms
Thread-25读取库存数据耗时：2495ms
Thread-6读取库存数据耗时：2498ms

结论：ReentrantReadWriteLock性能比synchronized快8倍

Condition接口有何用处？

同Object的 wait,notify/notifyAll 类似实现等待通知机制

 Condition接口和Lock配合来实现等待通知机制

Condition常用方法和使用范式

package com.lgs;

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

// Condition使用范式
public class ConditionTemplete {

    // 创建一个锁
    Lock lock = new ReentrantLock();
    
    // 从创建的锁里面拿到Condition
    Condition c = lock.newCondition();
    
    // 等待条件成熟
    public void waitCondition() throws InterruptedException {
        // 获取锁
        lock.lock();
        try {
            // 等待条件成熟
            c.await();
        }finally {
            // 释放锁
            lock.unlock();
        }
    }
    // 通知其他线程条件成熟
    public void waitNotify() throws InterruptedException {
        // 获取锁
        lock.lock();
        try {
            // 通知
            c.signal();
            // 尽量少使用signalAll() 因为condition是从lock创建出来的,要通知哪个线程是知道的
        }finally {
            // 释放锁
            lock.unlock();
        }
    }
}

结合ReentrantLock和Condition实现线程安全的有界队列

 类似于之前的wait和notify/notifyAll实现有界阻塞队列

package com.lgs.bq;

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

// Lock和Condition实现有界阻塞队列
// 类似于之前的wait和notify/notifyAll实现有界阻塞队列
public class BlockingQueueLockCondition<T> {

    // 定义一个list存放队列数据
    private List queue = new LinkedList<>();
    
    // 定义队列的容量
    private final int limit;
    
    // 定义Lock
    Lock lock = new ReentrantLock();
    
    // 创建不能为空的condition
    Condition notEmpty = lock.newCondition();
    
    // 创建不能满的condition
    Condition notFull = lock.newCondition();
    
    // 定义构造函数来初始化队列容量
    public BlockingQueueLockCondition(int limit) {
        this.limit = limit;
    } 
    
    // 入队
    public  void enqueue(T item) throws InterruptedException {
        // 获取锁
         lock.lock();
        try {
            // 当队列已经满了的时候就需要等待
            while (this.queue.size() == this.limit) {
                notFull.await();
            }
            // 数据入队
            this.queue.add(item);
            // 队列入队后就通知出队可以取数据了 因为可以肯定有出队的线程正在等待
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }
    
    // 出队
    public  T dequeue() throws InterruptedException {
        // 获取锁
        lock.lock();
        try {
            // 如果队列没有数据时就等待
            while (this.queue.size() == 0) {
                notEmpty.await();
            }
            // 数据即将出队就通知入队可以插入数据了 因为可以肯定有入队的线程正在等待
            notFull.signal();
            // 数据出队
            return (T) this.queue.remove(0);
        } finally {
            lock.unlock();
        }
    }
}

测试;

package com.lgs.bq;

// 测试等待通知机制实现的有界阻塞队列
public class BqTest {
    // 定义一个推数据入队列的线程
    private static class PushThread extends Thread{
        // 持有BlockingQueueLockCondition
        BlockingQueueLockCondition<Integer> bq;
        
        public PushThread(BlockingQueueLockCondition<Integer> bq) {
             this.bq = bq;
        }
        
        public void run() {
            int i = 20;
            while(i > 0) {
                System.out.println(" i=" + i +" will push");
                try {
                    Thread.sleep(500);
                    bq.enqueue(i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                i--;
            }
        }
    }
    
     //取数据出队列
    private static class PopThread extends Thread{
        BlockingQueueLockCondition<Integer> bq;

        public PopThread(BlockingQueueLockCondition<Integer> bq) {
            this.bq = bq;
        }
        @Override
        public void run() {
            while(true){
                try {
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getName()
                            +" will pop.....");
                    Integer i = bq.dequeue();
                    System.out.println(" i="+i.intValue()+" alread pop");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }
    }
    
    public static void main(String[] args) {
        // 创建一个有界阻塞队列的实例
        BlockingQueueLockCondition<Integer> bq = new BlockingQueueLockCondition(10);
        
        // 创建入队线程
        Thread t1 = new PushThread(bq);
        t1.setName("push");
        
        // 创建出队列线程
        Thread t2 = new PopThread(bq);
        t2.setName("pop");
        
        // 启动线程
        t1.start();
        t2.start();
    }
}

输出：

 i=20 will push
 i=19 will push
pop will pop.....
 i=20 alread pop
 i=18 will push
 i=17 will push
pop will pop.....
 i=19 alread pop
 i=16 will push
 i=15 will push
pop will pop.....
 i=18 alread pop
 i=14 will push
 i=13 will push
pop will pop.....
 i=17 alread pop
 i=12 will push
 i=11 will push
pop will pop.....
 i=16 alread pop
 i=10 will push
 i=9 will push
pop will pop.....
 i=15 alread pop
 i=8 will push
 i=7 will push
pop will pop.....
 i=14 alread pop
 i=6 will push
 i=5 will push
pop will pop.....
 i=13 alread pop
 i=4 will push
 i=3 will push
pop will pop.....
 i=12 alread pop
 i=2 will push
 i=1 will push
pop will pop.....
 i=11 alread pop
pop will pop.....
 i=10 alread pop
pop will pop.....
 i=9 alread pop
pop will pop.....
 i=8 alread pop
pop will pop.....
 i=7 alread pop
pop will pop.....
 i=6 alread pop
pop will pop.....
 i=5 alread pop
pop will pop.....
 i=4 alread pop
pop will pop.....
 i=3 alread pop
pop will pop.....
 i=2 alread pop
pop will pop.....
 i=1 alread pop
pop will pop.....