每日算法 — 使用java实现读者写者问题：读写锁同步与Java并发控制

可莉爸爸 AI算法

引言

读者-写者问题（Readers-Writers Problem）是操作系统课程中经典的进程同步问题，也是实际并发编程中频繁遇到的模式：多个线程可以同时读取共享资源，但写操作必须独占访问。本文将用 Java 从零实现读者优先、写者优先和公平策略三种版本的读写锁，并对比 java.util.concurrent.locks.ReentrantReadWriteLock 的工业级实现。

问题定义

约束条件

  • 读者之间不互斥：多个读者可同时访问共享资源
  • 写者之间互斥：同一时间只能有一个写者访问
  • 读者与写者互斥：写者访问时，任何读者都不能访问

三种策略

策略 特点 适用场景
读者优先 读者持续到来会饿死写者 读多写少，写操作不敏感
写者优先 写者持续到来会饿死读者 写操作要求高实时性
公平策略 按请求顺序排队，防止饿死 需要均衡读写延迟

读者优先实现

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

/**
 * 读者优先的读写锁实现
 */
public class ReaderPriorityReadWriteLock {

    private final Lock lock = new ReentrantLock();
    private final Condition canWrite = lock.newCondition();

    private int readerCount = 0;   // 当前活跃读者数
    private boolean isWriting = false; // 是否有写者正在写入

    /**
     * 读者获取读锁
     */
    public void readLock() throws InterruptedException {
        lock.lock();
        try {
            // 如果有写者正在写入，等待
            while (isWriting) {
                canWrite.await();
            }
            readerCount++;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 读者释放读锁
     */
    public void readUnlock() {
        lock.lock();
        try {
            readerCount--;
            // 最后一个读者离开，唤醒等待的写者
            if (readerCount == 0) {
                canWrite.signalAll();
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 写者获取写锁
     */
    public void writeLock() throws InterruptedException {
        lock.lock();
        try {
            // 必须等待所有读者离开且无其他写者
            while (readerCount > 0 || isWriting) {
                canWrite.await();
            }
            isWriting = true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 写者释放写锁
     */
    public void writeUnlock() {
        lock.lock();
        try {
            isWriting = false;
            canWrite.signalAll(); // 唤醒所有等待的读者和写者
        } finally {
            lock.unlock();
        }
    }
}

写者优先实现

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

/**
 * 写者优先的读写锁实现
 */
public class WriterPriorityReadWriteLock {

    private final Lock lock = new ReentrantLock();
    private final Condition canRead = lock.newCondition();
    private final Condition canWrite = lock.newCondition();

    private int readerCount = 0;
    private int waitingWriters = 0; // 等待的写者数量
    private boolean isWriting = false;

    public void readLock() throws InterruptedException {
        lock.lock();
        try {
            // 如果有写者正在写入或有写者在等待，读者等待
            while (isWriting || waitingWriters > 0) {
                canRead.await();
            }
            readerCount++;
        } finally {
            lock.unlock();
        }
    }

    public void readUnlock() {
        lock.lock();
        try {
            readerCount--;
            if (readerCount == 0) {
                canWrite.signal(); // 唤醒一个等待的写者
            }
        } finally {
            lock.unlock();
        }
    }

    public void writeLock() throws InterruptedException {
        lock.lock();
        try {
            waitingWriters++;
            // 等待所有读者离开且无其他写者
            while (readerCount > 0 || isWriting) {
                canWrite.await();
            }
            waitingWriters--;
            isWriting = true;
        } finally {
            lock.unlock();
        }
    }

    public void writeUnlock() {
        lock.lock();
        try {
            isWriting = false;
            // 优先唤醒等待的写者，其次唤醒读者
            if (waitingWriters > 0) {
                canWrite.signal();
            } else {
                canRead.signalAll();
            }
        } finally {
            lock.unlock();
        }
    }
}

公平策略实现

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

/**
 * 公平读写锁：按请求顺序依次服务，防止饿死
 */
public class FairReadWriteLock {

    private final Lock lock = new ReentrantLock(true); // 公平锁
    private final Condition condition = lock.newCondition();

    private int readerCount = 0;
    private int writerCount = 0; // 正在写入的写者数（0或1）
    private int waitingReaders = 0;
    private int waitingWriters = 0;

    // 使用一个标志记录当前模式：读模式、写模式、空闲
    private enum Mode { READING, WRITING, IDLE }
    private Mode mode = Mode.IDLE;

    public void readLock() throws InterruptedException {
        lock.lock();
        try {
            waitingReaders++;
            try {
                // 公平条件：只有空闲时才能进入，或已在读模式且无写者等待
                // 这里采用更严格的公平策略：如果前面有写者等待，读者必须等待
                while (mode == Mode.WRITING || waitingWriters > 0) {
                    condition.await();
                }
                readerCount++;
                mode = Mode.READING;
            } finally {
                waitingReaders--;
            }
        } finally {
            lock.unlock();
        }
    }

    public void readUnlock() {
        lock.lock();
        try {
            readerCount--;
            if (readerCount == 0) {
                mode = Mode.IDLE;
                condition.signalAll();
            }
        } finally {
            lock.unlock();
        }
    }

    public void writeLock() throws InterruptedException {
        lock.lock();
        try {
            waitingWriters++;
            try {
                while (mode != Mode.IDLE) {
                    condition.await();
                }
                writerCount = 1;
                mode = Mode.WRITING;
            } finally {
                waitingWriters--;
            }
        } finally {
            lock.unlock();
        }
    }

    public void writeUnlock() {
        lock.lock();
        try {
            writerCount = 0;
            mode = Mode.IDLE;
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }
}

基于信号量的简洁实现

import java.util.concurrent.Semaphore;

/**
 * 使用信号量实现的读写锁（读者优先）
 */
public class SemaphoreReadWriteLock {

    private final Semaphore resource = new Semaphore(1);   // 控制对资源的访问
    private final Semaphore mutex = new Semaphore(1);      // 保护 readerCount
    private int readerCount = 0;

    public void readLock() throws InterruptedException {
        mutex.acquire();
        readerCount++;
        if (readerCount == 1) {
            // 第一个读者获取资源锁
            resource.acquire();
        }
        mutex.release();
    }

    public void readUnlock() throws InterruptedException {
        mutex.acquire();
        readerCount--;
        if (readerCount == 0) {
            // 最后一个读者释放资源锁
            resource.release();
        }
        mutex.release();
    }

    public void writeLock() throws InterruptedException {
        resource.acquire();
    }

    public void writeUnlock() {
        resource.release();
    }
}

综合测试与性能对比

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockBenchmark {

    private static final int READERS = 10;
    private static final int WRITERS = 2;
    private static final int OPERATIONS = 100_000;

    private int sharedData = 0;

    public static void main(String[] args) throws Exception {
        // 测试自定义读者优先锁
        testCustomLock("ReaderPriority", new ReaderPriorityReadWriteLock());

        // 测试 JDK ReentrantReadWriteLock
        testJDKLock();
    }

    static void testCustomLock(String name, ReaderPriorityReadWriteLock rwLock) 
            throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(READERS + WRITERS);
        CountDownLatch latch = new CountDownLatch(READERS + WRITERS);
        AtomicInteger readCount = new AtomicInteger(0);
        AtomicInteger writeCount = new AtomicInteger(0);

        long start = System.currentTimeMillis();

        // 启动读者线程
        for (int i = 0; i < READERS; i++) {
            executor.submit(() -> {
                try {
                    for (int j = 0; j < OPERATIONS; j++) {
                        rwLock.readLock();
                        readCount.incrementAndGet();
                        rwLock.readUnlock();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    latch.countDown();
                }
            });
        }

        // 启动写者线程
        for (int i = 0; i < WRITERS; i++) {
            executor.submit(() -> {
                try {
                    for (int j = 0; j < OPERATIONS / 10; j++) {
                        rwLock.writeLock();
                        writeCount.incrementAndGet();
                        rwLock.writeUnlock();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    latch.countDown();
                }
            });
        }

        latch.await();
        long duration = System.currentTimeMillis() - start;

        System.out.println(name + " - 总耗时: " + duration + "ms");
        System.out.println("  读操作: " + readCount.get());
        System.out.println("  写操作: " + writeCount.get());

        executor.shutdown();
    }

    static void testJDKLock() throws InterruptedException {
        ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
        ExecutorService executor = Executors.newFixedThreadPool(READERS + WRITERS);
        CountDownLatch latch = new CountDownLatch(READERS + WRITERS);
        AtomicInteger readCount = new AtomicInteger(0);
        AtomicInteger writeCount = new AtomicInteger(0);

        long start = System.currentTimeMillis();

        for (int i = 0; i < READERS; i++) {
            executor.submit(() -> {
                for (int j = 0; j < OPERATIONS; j++) {
                    lock.readLock().lock();
                    readCount.incrementAndGet();
                    lock.readLock().unlock();
                }
                latch.countDown();
            });
        }

        for (int i = 0; i < WRITERS; i++) {
            executor.submit(() -> {
                for (int j = 0; j < OPERATIONS / 10; j++) {
                    lock.writeLock().lock();
                    writeCount.incrementAndGet();
                    lock.writeLock().unlock();
                }
                latch.countDown();
            });
        }

        latch.await();
        long duration = System.currentTimeMillis() - start;

        System.out.println("JDK ReentrantReadWriteLock - 总耗时: " + duration + "ms");
        System.out.println("  读操作: " + readCount.get());
        System.out.println("  写操作: " + writeCount.get());

        executor.shutdown();
    }
}

与 JDK ReentrantReadWriteLock 的对比

特性 自定义实现 JDK ReentrantReadWriteLock
锁降级 不支持 支持（写锁可降级为读锁）
可重入 不支持 支持
条件变量 不支持 支持（写锁可创建 Condition）
公平性 需手动实现 支持公平/非公平模式
性能 基础 经过 JVM 优化，性能更优
中断响应 支持 支持

实际应用场景

  • 缓存系统：高频读取、低频更新的缓存（如 Guava Cache、Caffeine）
  • 配置管理：应用配置读多写少，热更新时需要写锁
  • 元数据管理：数据库表的元信息访问
  • 路由表：网络路由表的高并发查询与偶发更新

总结

读者-写者问题是理解并发编程中”读共享、写独占”模式的经典案例。通过三种策略的实现对比，我们可以深入理解：

  1. 读者优先最大化读吞吐量，但可能饿死写者
  2. 写者优先保证写操作的低延迟，但可能饿死读者
  3. 公平策略平衡两者，但增加了线程切换开销

JDK 的 ReentrantReadWriteLock 在工业环境中经过充分验证，支持锁降级、可重入等高级特性，是生产环境的首选。理解其底层原理，有助于我们更好地进行并发系统的设计与调优。

Tagged with: