设计模式 Concurrency 之 ReadWriteLock 读写锁

定义

例子

1. 定义

这种模式允许同步的读操作,但是写操作会加排他锁.

意味着多线程可以进行读操作,但是修改数据的时候需要加排他锁

当一个线程在写数据时,其他读或写线程会一直阻塞到当前线程完成写操作

2. 例子

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, TimeUn
11b57
it.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