设计模式 Concurrency 之 ReadWriteLock 读写锁

深藏阁楼爱情的钟 2022-06-10 03:24 212阅读 0赞

*  定义
 *  例子

# 1. 定义 #

*  这种模式允许同步的读操作,但是写操作会加排他锁.
 *  意味着多线程可以进行读操作,但是修改数据的时候需要加排他锁
 *  当一个线程在写数据时,其他读或写线程会一直阻塞到当前线程完成写操作

# 2. 例子 #

![这里写图片描述][SouthEast]

ReadWriteLock

package com.hqq.concurrency.read_write_lock;
    
    import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReadWriteLock;
    
    /** * ReaderWriterLock * 读写锁 * 控制读写权限 * 允许多个读操作持有锁 但是如果加了写锁 读操作便会阻塞 * 如果加了读锁 写操作就会阻塞 * 是非公平锁 * Created by heqianqian on 2017/7/30. */
    public class ReaderWriterLock implements ReadWriteLock { 
    
    
        private Object readerMutex = new Object();
    
        private int currentReaderCount;
    
        private ReadLock readLock = new ReadLock();
    
        private WriteLock writeLock = new WriteLock();
    
        /** * 全局互斥变量 用来表示写操作和读操作是否同时获得锁 * <p> * 1.如果包含{@link readLock},表示当前线程持有读锁,其他线程仍然可以进行读操作 * 2.如果包含{@link writeLock},表示当前线程持有写锁,其他线程既无法执行写操作也无法执行读操作 */
        private Set<Object> globalMutex = new HashSet<>();
    
        @Override
        public Lock readLock() {
            return this.readLock;
        }
    
        @Override
        public Lock writeLock() {
            return this.writeLock;
        }
    
        private boolean doesWriterHoldLock() {
            return globalMutex.contains(writeLock);
        }
    
        private boolean doesReaderHoldLock() {
            return globalMutex.contains(readLock);
        }
    
    
        private boolean isLockFree() {
            return globalMutex.isEmpty();
        }
    
        private static void waitUnInterruptly(Object o) {
            try {
                o.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        /** * 读锁 */
        private class ReadLock implements Lock { 
    
            @Override
            public void lock() {
                synchronized (readerMutex) {
                    currentReaderCount++;//当前读操作数加1
                    if (currentReaderCount == 1) {
                        //尝试为第一个读线程获得互斥锁
                        synchronized (globalMutex) {
                            while (true) {
                                //如果没有线程持有锁或者当前持有的是读锁
                                if (isLockFree() || doesReaderHoldLock()) {
                                    globalMutex.add(this);
                                    break;
                                } else {
       //否则阻塞等待
                                    waitUnInterruptly(globalMutex);
                                }
                            }
                        }
                    }
                }
    
            }
    
            @Override
            public void lockInterruptibly() throws InterruptedException {
                throw new UnsupportedOperationException();
            }
    
            @Override
            public boolean tryLock() {
                throw new UnsupportedOperationException();
            }
    
            @Override
            public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
                throw new UnsupportedOperationException();
            }
    
            @Override
            public void unlock() {
                synchronized (readerMutex) {
                    //当前读线程数减1
                    currentReaderCount--;
                    //当读线程数为0[即完成所有读操作时 移除当前线程 唤醒其他等待线程]
                    if (currentReaderCount == 0) {
                        synchronized (globalMutex) {
                            globalMutex.remove(this);
                            //唤醒等待者 一般是读锁
                            globalMutex.notifyAll();
                        }
                    }
                }
            }
    
            @Override
            public Condition newCondition() {
                throw new UnsupportedOperationException();
            }
        }
    
        /** * 写锁 */
        private class WriteLock implements Lock { 
    
            @Override
            public void lock() {
                synchronized (globalMutex) {
                    while (true) {
                        if (isLockFree()) {
                            globalMutex.add(this);
                            break;
                        } else if (doesWriterHoldLock()) {
                            waitUnInterruptly(globalMutex);
                        } else if (doesReaderHoldLock()) {
                            waitUnInterruptly(globalMutex);
                        } else {
                            throw new AssertionError("it should never reach here");
                        }
                    }
                }
            }
    
            @Override
            public void lockInterruptibly() throws InterruptedException {
                throw new UnsupportedOperationException();
            }
    
            @Override
            public boolean tryLock() {
                throw new UnsupportedOperationException();
            }
    
            @Override
            public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
                throw new UnsupportedOperationException();
            }
    
            @Override
            public void unlock() {
                synchronized (globalMutex) {
                    globalMutex.remove(this);
                    globalMutex.notifyAll();
                }
            }
    
            @Override
            public Condition newCondition() {
                throw new UnsupportedOperationException();
            }
        }
    }

Reader

package com.hqq.concurrency.read_write_lock;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.locks.Lock;
    
    /** * Reader * Created by heqianqian on 2017/7/30. */
    public class Reader extends Thread { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(Reader.class);
    
        private String name;
    
        private Lock readLock;
    
        public Reader(String name, Lock readLock) {
            this.name = name;
            this.readLock = readLock;
        }
    
        @Override
        public void run() {
            readLock.lock();
            try {
                read();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                readLock.unlock();
            }
        }
    
        private void read() throws InterruptedException {
            LOGGER.info("{} read begin", name);
            Thread.sleep(500);
            LOGGER.info("{} read end", name);
        }
    }

Writer

package com.hqq.concurrency.read_write_lock;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.locks.Lock;
    
    /** * Writer * Created by heqianqian on 2017/7/30. */
    public class Writer extends Thread { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(Writer.class);
    
        private String name;
    
        private Lock writeLock;
    
        public Writer(String name, Lock writeLock) {
            this.name = name;
            this.writeLock = writeLock;
        }
    
        @Override
        public void run() {
            writeLock.lock();
            try {
                write();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                writeLock.unlock();
            }
        }
    
        public void write() throws InterruptedException {
            LOGGER.info("{} write begin", name);
            Thread.sleep(500);
            LOGGER.info("{} write end", name);
        }
    }

APP

package com.hqq.concurrency.read_write_lock;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.IntStream;
    
    /** * ReadWriteLock读写锁 * 在一个多线程的环境下,不管是读还是写操作会尝试去同步共享资源. * 这会导致低性能,特别在读多于写的情况下,因为读操作对另一个线程来说的线程安全的 * <p> * 这种模式允许同步的读操作,但是写操作会加排他锁. * 意味着多线程可以进行读操作,但是修改数据的时候需要加排他锁 * 当一个线程在写数据时,其他读或写线程会一直阻塞到当前线程完成写操作 * Created by heqianqian on 2017/7/30. */
    public class App { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(App.class);
    
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            ReaderWriterLock readerWriterLock = new ReaderWriterLock();
    
            //开启5个读线程
            IntStream.range(0, 5)
                    .forEach(i -> executorService.submit(new Reader("Reader" + i, readerWriterLock.readLock())));
    
            //开启5个写线程
            IntStream.range(0, 5)
                    .forEach(i -> executorService.submit(new Writer("Writer" + i, readerWriterLock.writeLock())));
    
            executorService.shutdown();
    
            try {
                executorService.awaitTermination(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                LOGGER.error("Error waiting for ExecutorService shutdown");
            }
        }
    
    }

测试结果:

INFO  [2017-08-09 02:24:34,339] com.hqq.concurrency.read_write_lock.Reader: Reader2 read begin
    INFO  [2017-08-09 02:24:34,336] com.hqq.concurrency.read_write_lock.Reader: Reader0 read begin
    INFO  [2017-08-09 02:24:34,337] com.hqq.concurrency.read_write_lock.Reader: Reader1 read begin
    INFO  [2017-08-09 02:24:34,352] com.hqq.concurrency.read_write_lock.Reader: Reader3 read begin
    INFO  [2017-08-09 02:24:34,361] com.hqq.concurrency.read_write_lock.Reader: Reader4 read begin
    INFO  [2017-08-09 02:24:34,849] com.hqq.concurrency.read_write_lock.Reader: Reader0 read end
    INFO  [2017-08-09 02:24:34,849] com.hqq.concurrency.read_write_lock.Reader: Reader1 read end
    INFO  [2017-08-09 02:24:34,849] com.hqq.concurrency.read_write_lock.Reader: Reader2 read end
    INFO  [2017-08-09 02:24:34,865] com.hqq.concurrency.read_write_lock.Reader: Reader3 read end
    INFO  [2017-08-09 02:24:34,866] com.hqq.concurrency.read_write_lock.Reader: Reader4 read end
    INFO  [2017-08-09 02:24:34,866] com.hqq.concurrency.read_write_lock.Writer: Writer4 write begin
    INFO  [2017-08-09 02:24:35,402] com.hqq.concurrency.read_write_lock.Writer: Writer4 write end
    INFO  [2017-08-09 02:24:35,402] com.hqq.concurrency.read_write_lock.Writer: Writer0 write begin
    INFO  [2017-08-09 02:24:35,904] com.hqq.concurrency.read_write_lock.Writer: Writer0 write end
    INFO  [2017-08-09 02:24:35,904] com.hqq.concurrency.read_write_lock.Writer: Writer3 write begin
    INFO  [2017-08-09 02:24:36,410] com.hqq.concurrency.read_write_lock.Writer: Writer3 write end
    INFO  [2017-08-09 02:24:36,411] com.hqq.concurrency.read_write_lock.Writer: Writer1 write begin
    INFO  [2017-08-09 02:24:36,911] com.hqq.concurrency.read_write_lock.Writer: Writer1 write end
    INFO  [2017-08-09 02:24:36,912] com.hqq.concurrency.read_write_lock.Writer: Writer2 write begin
    INFO  [2017-08-09 02:24:37,415] com.hqq.concurrency.read_write_lock.Writer: Writer2 write end

--------------------

[SouthEast]: /images/20220610/922d9549650d454ea5704269a2994c72.png