您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 多執行緒檔案讀寫佇列實作：生產者消費者模式

編輯：Java 相關知識

高效能檔案寫入架構：基於 Lock/Condition 的執行緒安全佇列

本範例實作一個執行緒安全的檔案寫入佇列，採用單例模式管理共享佇列，使用 ReentrantLock 與 Condition 實現生產者-消費者模式。適用於高併發日誌寫入、批次資料持久化等場景，支援背壓控制避免記憶體溢位。

架構設計重點

單例模式：雙重檢查鎖定（靜態內部類實現）確保全域唯一佇列實例
有界佇列：固定容量（預設 10,000）防止無限制堆積導致 OOM
條件變數：full（佇列滿等待）、empty（佇列空等待）實現精確喚醒
批次操作：支援單筆 add() 與批次 addAll() 提升吞吐量

完整程式碼實作

package com.ysh.file;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.google.common.collect.Lists;

/**
 * 執行緒安全檔案寫入佇列（生產者-消費者模式）
 */
public class WriteQueue {
    
    private static final int MAX_QUEUE_SIZE = 10_000;
    
    // 使用 LinkedList 實現高效首尾操作
    private final LinkedList queueCache = Lists.newLinkedList();
    
    // 重入鎖 + 條件變數
    private final Lock lock = new ReentrantLock();
    private final Condition full = lock.newCondition();   // 佇列滿等待
    private final Condition empty = lock.newCondition();  // 佇列空等待

    // 單例模式：靜態內部類延遲初始化（執行緒安全）
    private WriteQueue() {}
    
    private static class WriteQueueHolder {
        private static final WriteQueue INSTANCE = new WriteQueue();
    }
    
    public static WriteQueue getInstance() {
        return WriteQueueHolder.INSTANCE;
    }

    /**
     * 生產者：加入單筆資料（阻塞直到有空間）
     */
    public void add(String line) {
        lock.lock();
        try {
            // 佇列滿時等待
            while (queueCache.size() >= MAX_QUEUE_SIZE) {
                full.await();
            }
            queueCache.addFirst(line);
            // 喚醒消費者
            empty.signal();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 保留中斷狀態
            throw new RuntimeException("寫入佇列被中斷", e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 生產者：批次加入資料（阻塞直到有足夠空間）
     */
    public void add(List lineList) {
        lock.lock();
        try {
            while (queueCache.size() >= MAX_QUEUE_SIZE) {
                full.await();
            }
            queueCache.addAll(lineList);
            empty.signal();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("批次寫入佇列被中斷", e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 消費者：取出所有資料並清空佇列（阻塞直到有資料）
     * @return 包含所有佇列資料的 ArrayList
     */
    public ArrayList flush() {
        ArrayList temp = null;
        lock.lock();
        try {
            // 佇列空時等待
            while (queueCache.size() == 0) {
                empty.await();
            }
            temp = Lists.newArrayList();
            temp.addAll(queueCache);
            queueCache.clear();
            // 喚醒生產者
            full.signal();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("讀取佇列被中斷", e);
        } finally {
            lock.unlock();
        }
        return temp;
    }
}

使用範例：生產者消費者協作

// 生產者執行緒（模擬日誌產生）
new Thread(() -> {
    WriteQueue queue = WriteQueue.getInstance();
    for (int i = 0; i < 1000; i++) {
        queue.add("LOG-" + System.currentTimeMillis() + "-訊息內容" + i);
        // 模擬間歇性產生
        if (i % 100 == 0) { try { Thread.sleep(10); } catch (InterruptedException ignored) {} }
    }
}, "Log-Producer").start();

// 消費者執行緒（模擬檔案寫入）
new Thread(() -> {
    WriteQueue queue = WriteQueue.getInstance();
    while (true) {
        ArrayList batch = queue.flush();
        // 實際寫入檔案邏輯
        writeToFile(batch);
        if (batch.isEmpty()) break; // 結束條件
    }
}, "File-Consumer").start();