首先看一下JUC的重磅武器——锁（Lock）
相比同步锁，JUC包中的Lock锁的功能更加强大，它提供了各种各样的锁（公平锁，非公平锁，共享锁，独占锁……），所以使用起来很灵活。
翻译过来就是：
锁实现提供了比使用同步方法和语句可以获得的更广泛的锁操作。它们允许更灵活的结构，可能具有非常不同的属性，并且可能支持多个关联的条件对象。
Lock是一个接口，这里主要有三个实现：
- ReentrantLock
- ReentrantReadWriteLock.ReadLock
- ReentrantReadWriteLock.WriteLock
一、ReentrantLock可重入锁(递归锁)
使用ReentrantLock改造卖票程序：只需改造sale()方法
private ReentrantLock lock = new ReentrantLock(); //创建实例对象
lock.lock(); //加锁
lock.unlock(); //释放锁
class Ticket{private Integer number = 20;private ReentrantLock lock = new ReentrantLock();public void sale(){lock.lock();if (number <= 0) {System.out.println("票已售罄！");lock.unlock();return;}try {Thread.sleep(200);number--;System.out.println(Thread.currentThread().getName() + "买票成功，当前剩余：" + number);} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}
}1. 测试可重入性
可重入锁又名递归锁，是指在同一个线程在外层方法获取锁的时候，再进入该线程的内层方法会自动获取锁。Java中ReentrantLock和synchronized都是可重入锁，可重入锁的一个优点是可一定程度避免死锁。
例如下列伪代码：
class A{public synchronized void aa{......bb();......}public synchronized void bb{......}
}
A a = new A();
a.aa();A类中有两个普通同步方法，都需要对象a的锁。如果是不可重入锁的话，aa方法首先获取到锁，aa方法在执行的过程中需要调用bb方法，此时锁被aa方法占有，bb方法无法获取到锁，这样就会导致bb方法无法执行，aa方法也无法执行，出现了死锁情况。可重入锁可避免这种死锁的发生。
class Ticket{private Integer number = 20;private ReentrantLock lock = new ReentrantLock();public void sale(){lock.lock();if (number <= 0) {System.out.println("票已售罄！");lock.unlock();return;}try {Thread.sleep(200);number--;System.out.println(Thread.currentThread().getName() + "买票成功，当前剩余：" + number);// 调用check方法测试锁的可重入性this.check();} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}/*** 为了测试可重入锁，添加检查余票方法*/public void check(){lock.lock();System.out.println("检查余票。。。。");lock.unlock();}
}可以发现程序可以正常执行。。。说明该锁确实可重入。
AAA买票成功，当前剩余：19
检查余票。。。。
AAA买票成功，当前剩余：18
检查余票。。。。
AAA买票成功，当前剩余：17
检查余票。。。。
AAA买票成功，当前剩余：16
检查余票。。。。
AAA买票成功，当前剩余：15
检查余票。。。。
AAA买票成功，当前剩余：14
检查余票。。。。
AAA买票成功，当前剩余：13
检查余票。。。。
BBB买票成功，当前剩余：12
检查余票。。。。
BBB买票成功，当前剩余：11
检查余票。。。。
BBB买票成功，当前剩余：10
。。。。。。
2. 测试公平锁
ReentrantLock还可以实现公平锁。所谓公平锁，也就是在锁上等待时间最长的线程将获得锁的使用权。通俗的理解就是谁排队时间最长谁先执行获取锁。
ReentrantLock lock = new ReentrantLock(); //默认非公平锁
ReentrantLock lock = new ReentrantLock(true); //true表示创建公平锁
//默认非公平锁//ReentrantLock lock = new ReentrantLock();//true表示创建公平锁ReentrantLock lock = new ReentrantLock(true);public void test() throws InterruptedException {//lock.tryLock():获取锁并立即返货获取锁的结果，成功返回true，失败false
// lock.tryLock(timeout,timeunit): 最多阻塞等待timeout单位timeunit 时间，获取成功返回true，失败falseif(lock.tryLock(6, TimeUnit.SECONDS)){//获取锁成功System.out.println(Thread.currentThread().getName()+"开始执行...");Thread.sleep(200);System.out.println(Thread.currentThread().getName()+"执行结束...");lock.unlock();}else{System.out.println(Thread.currentThread().getName()+"获取锁失败....");}}
}测试结果：可以看到ABC三个线程是按顺序买票成功的。
AAA买票成功，当前剩余：19
检查余票。。。。
BBB买票成功，当前剩余：18
检查余票。。。。
CCC买票成功，当前剩余：17
检查余票。。。。
AAA买票成功，当前剩余：16
检查余票。。。。
BBB买票成功，当前剩余：15
检查余票。。。。
CCC买票成功，当前剩余：14
。。。。。。
3. 限时等待
这个是什么意思呢？也就是通过我们的tryLock方法来实现，可以选择传入时间参数，表示等待指定的时间，无参则表示立即返回锁申请的结果：true表示获取锁成功，false表示获取锁失败。我们可以将这种方法用来解决死锁问题。
lock.tryLock():获取锁并立即返货获取锁的结果，成功返回true，失败false
lock.tryLock(timeout,timeunit): 最多阻塞等待timeout单位timeunit 时间，获取成功返回true，失败false
public class Demo3 {public static void main(String[] args) {Demo3 demo3 = new Demo3();new Thread(()->{demo3.test();},"A").start();new Thread(()->{demo3.test();},"B").start();}Lock lock = new ReentrantLock();public void test(){try {boolean b = lock.tryLock(1000, TimeUnit.MILLISECONDS);if(!b){System.out.println(Thread.currentThread().getName()+"获取锁失败");return;}Thread.sleep(2000);System.out.println(Thread.currentThread().getName()+"....");lock.unlock();} catch (InterruptedException e) {e.printStackTrace();lock.unlock();}}
}4. ReentrantLock和synchronized区别
（1）synchronized是独占锁，加锁和解锁的过程自动进行，易于操作，但不够灵活。ReentrantLock也是独占锁，加锁和解锁的过程需要手动进行，不易操作，但非常灵活。
（2）synchronized可重入，因为加锁和解锁自动进行，不必担心最后是否释放锁；ReentrantLock也可重入，但加锁和解锁需要手动进行，且次数需一样，否则其他线程无法获得锁。
（3）synchronized不可响应中断，一个线程获取不到锁就一直等着；ReentrantLock可以响应中断。
（4）synchronzied锁的是对象，锁是保存在对象头里面的，根据对象头数据来标识是否有线程获得锁/争抢锁；ReentrantLock锁的是线程，根据进入的线程和int类型的state标识锁的获得/争抢。
二、ReentrantReadWriteLock读写锁
在并发场景中用于解决线程安全的问题，我们几乎会高频率的使用到独占式锁，通常使用java提供的关键字synchronized或者concurrents包中实现了Lock接口的ReentrantLock。它们都是独占式获取锁，也就是在同一时刻只有一个线程能够获取锁。
现实中有这样一种场景：对共享资源有读和写的操作，且写操作没有读操作那么频繁。在没有写操作的时候，多个线程同时读一个资源没有任何问题，所以应该允许多个线程同时读取共享资源；但是如果一个线程想去写这些共享资源，就不应该允许其他线程对该资源进行读和写的操作了。大部分只是读数据，写数据很少，如果仅仅是读数据的话并不会影响数据正确性（出现脏读），而如果在这种业务场景下，依然使用独占锁的话，很显然这将是出现性能瓶颈的地方。针对这种读多写少的情况，java还提供了另外一个实现Lock接口的ReentrantReadWriteLock(读写锁)。读写锁允许同一时刻被多个读线程访问，但是在写线程访问时，所有的读线程和其他的写线程都会被阻塞。
接下来以缓存为例用代码演示读写锁，重现问题：
class MyCache{private volatile Map<String, String> cache= new HashMap<>();public void put(String key, String value){try {System.out.println(Thread.currentThread().getName() + " 开始写入！");Thread.sleep(300);cache.put(key, value);System.out.println(Thread.currentThread().getName() + " 写入成功！");} catch (InterruptedException e) {e.printStackTrace();} finally {}}public void get(String key){try {System.out.println(Thread.currentThread().getName() + " 开始读出！");Thread.sleep(300);String value = cache.get(key);System.out.println(Thread.currentThread().getName() + " 读出成功！" + value);} catch (InterruptedException e) {e.printStackTrace();} finally {}}
}public class ReentrantReadWriteLockDemo {public static void main(String[] args) {MyCache cache = new MyCache();for (int i = 1; i <= 5; i++) {String num = String.valueOf(i);// 开启5个写线程new Thread(()->{cache.put(num, num);}, num).start();}for (int i = 1; i <= 5; i++) {String num = String.valueOf(i);// 开启5个读线程new Thread(()->{cache.get(num);}, num).start();}}
}打印结果：多执行几次，有很大概率不会出现问题
改造MyCache，加入读写锁
class MyCache{private volatile Map<String, String> cache= new HashMap<>();// 加入读写锁ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();public void put(String key, String value){// 加写锁rwl.writeLock().lock();try {System.out.println(Thread.currentThread().getName() + " 开始写入！");Thread.sleep(500);cache.put(key, value);System.out.println(Thread.currentThread().getName() + " 写入成功！");} catch (InterruptedException e) {e.printStackTrace();} finally {// 释放写锁rwl.writeLock().unlock();}}public void get(String key){// 加入读锁rwl.readLock().lock();try {System.out.println(Thread.currentThread().getName() + " 开始读出！");Thread.sleep(500);String value = cache.get(key);System.out.println(Thread.currentThread().getName() + " 读出成功！" + value);} catch (InterruptedException e) {e.printStackTrace();} finally {// 释放读锁rwl.readLock().unlock();}}
}
![流批一体计算引擎-4-[Flink]消费kafka实时数据](https://img-blog.csdnimg.cn/2a99812e21b5428fb74bb88356306047.png)