java并发 - 学习ReentrantLock

太过爱你忘了你带给我的痛 2021-09-29 10:08 305阅读 0赞

**学习ReentrantLock**

**一丶什么是ReentrantLock**

**　　**ReentrantLock是java并发包中锁的一种实现, 它是重入锁, 即获取锁之后,自身可再次获取锁, 自身并不会造成死锁, 可以认为它是关键字sychronized的替代品,但它比sychronized有更丰富的功能, 如提供Condition等待, 提供中断等功能.

**二丶ReentrantLock的功能**

ReentrantLock实现了Lock接口, 因此它有如下功能

1. 获取锁

void lock();

2. 获取锁, 等待过程可中断

void lockInterruptibly() throws InterruptedException;

3. 尝试获取锁, 如果没有获取到锁, 立即返回

boolean tryLock();

4. 尝试获取锁, 并且在指定时间内等待, 当超出指定时间后, 或被中断之后立即返回

boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

5. 释放锁

void unlock();

6. 新建条件

Condition newCondition();

**三丶使用例子**

使用ReentrantLock成为线程安全的阻塞队列

public class BlockingQueue {
        
        private int[] datas;
        private int count;
        private int firstIndex;
        private int endIndex;
        
        private ReentrantLock lock=new ReentrantLock();
        
        // 未空的条件， 等待take
        private final Condition notEmpty = lock.newCondition();
        // 未满的条件， 等待put
        private final Condition notFull = lock.newCondition();
        
        
        public BlockingQueue(int capacity) {
            this.datas = new int[capacity];
            this.count = 0;
        }
        
        public void put(int data) throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly(); // 先获取锁 
            try {
                while(count == datas.length){ //这里需要加上循环判断, 队列已满, 不能存, 释放锁, 挂起等待
                    notFull.await();  // 如果条件不满足, 会先释放锁, 挂起当前线程, 直到条件满足被通知唤醒
                }
                enqueue(data);
            }finally {
                lock.unlock();
            }
        }
        
        public int take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly(); // 先获取锁 
            try {
                while(count == 0){  // 这里需要加上循环判断, 个数为0, 不能取, 释放锁, 挂起等待
                    notEmpty.await(); // 如果条件不满足, 会先释放锁, 挂起当前线程, 直到条件满足被通知唤醒
                }
                int data=dequeue();
                return data;
            }finally {
                lock.unlock();
            }
        }
        
        private void enqueue(int data){
            datas[endIndex]=data;
            
            
            endIndex++;
            if(endIndex==datas.length){
                endIndex=0;
            }
            
            count++;
            notEmpty.signal();  //condition 发信号，将内部队列的第一个线程移至Lock队列中
        }
        
        private int dequeue(){
            int data=datas[firstIndex];
         
            firstIndex++;
            if(firstIndex==datas.length){
                firstIndex=0;
            }
            
            count--;
            notFull.signal();
            return data;
        }
        
    }

场景测试类:

public class ReentrantTests {
    
        public static void main(String[] args) throws InterruptedException {
            // 
            BlockingQueue blockingQueue=new BlockingQueue(20);
            
            Thread producer=new Thread(new ProducerRunnable(blockingQueue, 1));
            Thread consumer1=new Thread(new ConsumerRunnable("消费者1", blockingQueue, 2));
            Thread consumer2=new Thread(new ConsumerRunnable("消费者2", blockingQueue, 3));
            
            
            producer.start();
            consumer1.start();
            consumer2.start();
            
    //        producer.join();
            consumer1.join();
            consumer2.join();
        }
        
        
        
    }
    class ProducerRunnable implements  Runnable{
    
        private BlockingQueue bq;
        private int interval;
        
        public ProducerRunnable(BlockingQueue bq, int interval) {
            this.bq = bq;
            this.interval=interval;
        }
    
        @Override public void run() {
            int mix=10000;
            int i=0;
            try {
                while(i<mix){
                    this.bq.put(i);
                    i++;
                    System.out.println("生产者停顿"+interval*100+"ms...");
                    Thread.sleep(interval * 100);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    
    class ConsumerRunnable implements  Runnable{
    
        private BlockingQueue bq;
        private int interval;
        private String name;
    
        public ConsumerRunnable(String name,BlockingQueue bq, int interval) {
            this.name = name;
            this.bq = bq;
            this.interval=interval;
        }
    
        @Override public void run() {
            int mix=10000;
            int i=0;
            try {
                while(i<mix){
                    int data=this.bq.take();
                    System.out.println(this.name+"获取数据"+data);
                    System.out.println(this.name+"停顿"+interval*100+"ms...");
                    Thread.sleep(interval * 100);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

完整源码: [ReentraLock][]

转载于:https://www.cnblogs.com/timfruit/p/10962870.html

[ReentraLock]: https://gitee.com/timfruit189/java-learning/tree/master/src/main/java/com/ttx/java/concurrent/aqs