Java多线程同步：解决线程饥饿问题的策略与实践-java教程-PHP中文网

Java多线程同步：解决线程饥饿问题的策略与实践

本文探讨在java多线程环境中，如何通过不同策略解决锁竞争导致的线程饥饿问题。从简单的线程休眠到随机休眠，再到更高级的`wait()`和`notifyall()`机制，文章详细分析了各种方法的适用场景、优缺点及其在确保线程公平性方面的作用，旨在提供一套全面的多线程同步实践指南。

在多线程编程中，当多个线程尝试访问共享资源并需要获取锁时，可能会出现线程饥饿（Thread Starvation）问题。线程饥饿是指某个线程或一组线程由于优先级、调度策略或其他原因，长时间无法获取到所需的资源或执行机会，从而导致其任务无法完成。尤其当存在一个“无限循环”的线程频繁获取并释放锁时，其他等待获取相同锁的线程可能会长时间得不到执行。

线程饥饿问题的初步应对：引入休眠

为了缓解一个高频次执行的线程（例如在一个无限循环中运行的线程）对锁的持续占用，一种直观的策略是在该线程释放锁后引入短暂的休眠。这种做法的目的是让出CPU时间片，给其他等待锁的线程提供获取锁的机会。

考虑以下场景，一个线程在一个无限循环中尝试获取一个ReentrantLock，执行一些操作后释放锁，然后休眠片刻：

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

public class LockContentionExample {
    private final Lock writeLock = new ReentrantLock(true); // 使用公平锁

    public void infiniteLoopTask() {
        while (true) {
            try {
                // 尝试获取锁，设置超时时间防止无限等待
                if (writeLock.tryLock(100, TimeUnit.MILLISECONDS)) {
                    try {
                        System.out.println(Thread.currentThread().getName() + " acquired lock and doing something.");
                        // 模拟业务操作
                        Thread.sleep(50); 
                    } finally {
                        writeLock.unlock();
                        System.out.println(Thread.currentThread().getName() + " released lock.");
                    }
                } else {
                    System.out.println(Thread.currentThread().getName() + " failed to acquire lock, retrying...");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println(Thread.currentThread().getName() + " interrupted.");
                break;
            }

            // 关键点：释放锁后休眠，给其他线程机会
            try {
                Thread.sleep(10); // 固定休眠时间
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println(Thread.currentThread().getName() + " interrupted during sleep.");
                break;
            }
        }
    }

    public static void main(String[] args) {
        LockContentionExample example = new LockContentionExample();
        new Thread(example::infiniteLoopTask, "InfiniteLoopThread").start();

        // 模拟一个调度任务线程
        new Thread(() -> {
            while (true) {
                try {
                    if (example.writeLock.tryLock(100, TimeUnit.MILLISECONDS)) {
                        try {
                            System.out.println(Thread.currentThread().getName() + " acquired lock for scheduled task.");
                            Thread.sleep(20); 
                        } finally {
                            example.writeLock.unlock();
                            System.out.println(Thread.currentThread().getName() + " released lock after scheduled task.");
                        }
                    } else {
                        System.out.println(Thread.currentThread().getName() + " failed to acquire lock for scheduled task, retrying...");
                    }
                    Thread.sleep(500); // 调度任务间隔
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }, "ScheduledTaskThread").start();

        // 模拟另一个手动触发的任务线程
        new Thread(() -> {
            try {
                Thread.sleep(1500); // 延迟启动
                if (example.writeLock.tryLock(500, TimeUnit.MILLISECONDS)) {
                    try {
                        System.out.println(Thread.currentThread().getName() + " acquired lock for manual task.");
                        Thread.sleep(100); 
                    } finally {
                        example.writeLock.unlock();
                        System.out.println(Thread.currentThread().getName() + " released lock after manual task.");
                    }
                } else {
                    System.out.println(Thread.currentThread().getName() + " failed to acquire lock for manual task.");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "ManualTaskThread").start();
    }
}

登录后复制

在上述代码中，InfiniteLoopThread在每次释放锁后休眠10毫秒。对于只有两个线程（一个无限循环线程和一个调度任务线程）竞争锁的简单场景，这种固定休眠时间通常能够有效缓解饥饿问题，因为休眠提供了一个可预测的窗口，允许另一个线程尝试获取锁。

立即学习“Java免费学习笔记（深入）”；

多线程竞争下的优化：引入随机休眠

当竞争锁的线程数量增加到三个或更多时，固定休眠时间可能会再次导致饥饿。例如，如果线程A（无限循环）释放锁，线程B和线程C都在等待。如果线程B总是比线程C更快地尝试获取锁，并且在线程A释放锁的瞬间成功获取，那么线程C可能仍然会长时间无法获取到锁。这种情况下，调度变得可预测，导致某些线程持续被“跳过”。

为了打破这种可预测性，可以引入随机休眠时间。通过让线程休眠一个随机的时长（例如，在5到100毫秒之间），可以增加所有等待线程获取锁的机会，从而提高公平性。

豆包AI编程

豆包推出的AI编程助手

483

查看详情

// ... (代码其余部分与上例相同)

// 关键点：释放锁后休眠，引入随机性
try {
    // 假设 RandomUtil.nextInt(min, max) 返回一个 min 到 max-1 之间的随机整数
    // 这里使用 Java 标准库的 Random 类模拟
    int randomSleepTime = new java.util.Random().nextInt(96) + 5; // 5到100毫秒
    Thread.sleep(randomSleepTime); 
    System.out.println(Thread.currentThread().getName() + " sleeping for " + randomSleepTime + "ms (random).");
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    System.err.println(Thread.currentThread().getName() + " interrupted during random sleep.");
    break;
}

// ... (代码其余部分与上例相同)

登录后复制

引入随机休眠的好处在于，它使得线程B和线程C尝试获取锁的时机变得不确定，从而增加了每个线程获得锁的概率，减少了特定线程持续饥饿的可能性。然而，无论是固定休眠还是随机休眠，都存在性能开销，因为线程在休眠期间不执行任何有用的工作。

更健壮的解决方案：wait() 和 notifyAll()

尽管休眠策略在某些简单场景下有效，但它本质上是一种基于“忙等待”（polling）和猜测的机制。更推荐和健壮的解决方案是使用Java提供的Object.wait()和Object.notifyAll()机制，这是一种基于协作的线程通信方式，能够更高效、更公平地处理线程间的资源竞争。

wait()和notifyAll()通常与synchronized关键字配合使用，它们依赖于对象的内部监视器（monitor）。当一个线程调用wait()时，它会释放当前持有的监视器锁，并进入等待状态，直到被其他线程notify()或notifyAll()唤醒。

以下是一个使用wait()和notifyAll()来协调线程访问共享资源的示例。在这个模型中，不再是线程忙于尝试获取锁，而是当资源不可用时线程进入等待状态，当资源可用时被通知唤醒。

import java.util.Queue;
import java.util.LinkedList;

public class WaitNotifyExample {
    private final Queue<Integer> sharedQueue = new LinkedList<>();
    private final int CAPACITY = 5; // 队列容量

    // 生产者任务：模拟无限循环线程，生产数据并放入队列
    public void producerTask() {
        int item = 0;
        while (true) {
            synchronized (sharedQueue) {
                // 如果队列已满，生产者等待
                while (sharedQueue.size() == CAPACITY) {
                    try {
                        System.out.println("Producer: Queue is full, waiting...");
                        sharedQueue.wait(); // 释放锁并等待
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                }

                // 队列不满，生产数据
                sharedQueue.add(item);
                System.out.println("Producer produced: " + item);
                item++;

                sharedQueue.notifyAll(); // 通知所有等待的消费者
            }
            try {
                Thread.sleep(100); // 模拟生产间隔
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }
    }

    // 消费者任务：模拟调度任务线程或手动触发任务线程，从队列消费数据
    public void consumerTask(String name) {
        while (true) {
            synchronized (sharedQueue) {
                // 如果队列为空，消费者等待
                while (sharedQueue.isEmpty()) {
                    try {
                        System.out.println(name + ": Queue is empty, waiting...");
                        sharedQueue.wait(); // 释放锁并等待
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                }

                // 队列不空，消费数据
                int consumedItem = sharedQueue.remove();
                System.out.println(name + " consumed: " + consumedItem);

                sharedQueue.notifyAll(); // 通知所有等待的生产者/其他消费者
            }
            try {
                Thread.sleep(200); // 模拟消费间隔
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
        }
    }

    public static void main(String[] args) {
        WaitNotifyExample example = new WaitNotifyExample();

        new Thread(example::producerTask, "ProducerThread").start();
        new Thread(() -> example.consumerTask("ConsumerThread-1"), "ConsumerThread-1").start();
        new Thread(() -> example.consumerTask("ConsumerThread-2"), "ConsumerThread-2").start();
    }
}

登录后复制

在wait()和notifyAll()模型中：

线程不再通过忙等待或周期性休眠来猜测锁是否可用。
当一个线程需要某个条件（例如，队列非空或非满）才能继续执行时，它调用wait()，从而释放锁并进入等待状态，直到条件满足并被notify()或notifyAll()唤醒。
当另一个线程改变了条件（例如，向队列添加了元素或移除了元素）后，它调用notifyAll()来唤醒所有等待在该对象上的线程，这些被唤醒的线程将重新竞争锁并检查条件。
JVM在调度notifyAll()唤醒的线程时，会尝试提供更公平的调度，从而减少饥饿的风险。

注意事项与总结

锁的类型选择： ReentrantLock提供了比synchronized更灵活的功能，如公平锁（new ReentrantLock(true)），它会尽量保证等待时间最长的线程优先获取锁，但公平锁会带来一定的性能开销。wait()/notifyAll()则与synchronized关键字（即对象的内部监视器）紧密绑定。在选择同步机制时，需根据具体需求权衡。
try-finally的重要性： 无论使用ReentrantLock还是synchronized，务必在finally块中释放锁或退出同步块，以防止因异常导致锁无法释放，造成死锁。
wait()的条件判断： 始终在循环中调用wait()，即while (condition)而不是if (condition)。这是因为线程可能被虚假唤醒（spurious wakeup），或者在被唤醒后，条件可能再次变得不满足。
notify() vs notifyAll()： notify()只唤醒一个等待线程，而notifyAll()唤醒所有等待线程。在不确定哪个线程应该被唤醒的情况下，或者当有多种类型的线程等待时，使用notifyAll()更安全，以避免饥饿。
性能考量： 线程休眠虽然简单，但会造成CPU资源的浪费，因为线程在休眠期间不执行任何有效工作。wait()/notifyAll()机制通过让线程进入等待状态，减少了CPU的忙等待，通常更为高效。
更高级的并发工具： Java并发包（java.util.concurrent）提供了更高级、更易用的并发工具，如Semaphore（信号量）、CountDownLatch（倒计时锁）、CyclicBarrier（循环屏障）以及BlockingQueue（阻塞队列）等，它们可以帮助开发者更优雅地解决复杂的线程同步问题，避免手动实现wait()/notifyAll()的复杂性。例如，上述生产者-消费者模型可以直接使用ArrayBlockingQueue等阻塞队列来实现，代码会更加简洁和健壮。