Java 多執行緒非同步建立 Solr 索引：生產者-消費者模式與批次提交優化

架構設計

採用 單一背景執行緒 + 阻塞佇列 模式：

• 前端執行緒呼叫 addIndex()/delIndex() 將操作封裝為 Operation 丟入 ArrayBlockingQueue（非阻塞 offer）。
• 專屬 UpdateTask 執行緒輪詢佇列，累積 1 秒或佇列滿時批次提交 Solr，大幅降低 HTTP 請求數。
• 優雅關機：送入 StopOperation，等待佇列清空再關閉客戶端。

核心類別結構

public class SolrIndexer implements Indexer, Searcher, DisposableBean {
    private final CommonsHttpSolrServer server;
    private final BlockingQueue<Operation> inputQueue = new ArrayBlockingQueue<>(16384);
    private final Thread updateThread;
    private volatile boolean running = true, shuttingDown = false;

    public SolrIndexer(String url) throws MalformedURLException {
        server = new CommonsHttpSolrServer(url);
        updateThread = new Thread(new UpdateTask(), "SolrIndexer");
        updateThread.start();
    }

    // 生產者方法
    public void addIndex(Indexable idx) { inputQueue.offer(new Operation(idx, OperationType.UPDATE)); }
    public void delIndex(Indexable idx) { inputQueue.offer(new Operation(idx, OperationType.DELETE)); }

    // 消費者任務
    class UpdateTask implements Runnable {
        public void run() {
            while (running) {
                try { syncIndices(); }
                catch (Throwable e) { log.error("索引更新異常", e); }
            }
        }
    }

    private void syncIndices() throws InterruptedException {
        Operation op = inputQueue.poll(1000, TimeUnit.MILLISECONDS);
        if (op == null) return;
        if (op instanceof StopOperation) { ((StopOperation)op).stop(); return; }

        Thread.sleep(1000); // 等待更多操作湊成批次
        List<Operation> ops = new ArrayList<>(inputQueue.size() + 1);
        ops.add(op);
        inputQueue.drainTo(ops);

        // 按類型與操作分組
        Map<String, List<Indexable>> updateMap = new HashMap<>(), deleteMap = new HashMap<>();
        for (Operation o : ops) {
            Map<String, List<Indexable>> map = o.type == OperationType.DELETE ? deleteMap : updateMap;
            map.computeIfAbsent(o.indexable.getType(), k -> new LinkedList<>()).add(o.indexable);
        }

        // 批次提交
        deleteMap.forEach(this::delIndices);
        updateMap.forEach(this::updateIndices);
    }

    // 實際 Solr 交互略...
}

關鍵優化點

SolrJ 批次提交關鍵代碼

private void updateIndices(String type, List<Indexable> list) {
    UpdateRequest req = new UpdateRequest("/" + type + "/update");
    req.setAction(UpdateRequest.ACTION.COMMIT, false, false); // 軟提交
    for (Indexable idx : list) {
        SolrInputDocument doc = new SolrInputDocument();
        idx.getDoc().forEach(f -> doc.addField(f.getName(), f.getValue(), f.getBoost()));
        req.add(doc);
    }
    req.process(server);
}

擴展建議

• 多 Shard 路由：根據 Indexable.getType() 或 ID 雜湊分發至不同 Collection。
• 重試機制：網路抖動時指數退避重試，失敗進入死信佇列報警。
• 監控指標：佇列長度、批次大小、提交延遲、失敗率接入 Prometheus/Grafana。
• SolrCloud：生產環境建議使用 SolrCloud + ZooKeeper，客戶端改用 CloudSolrClient 自動發現 Leader。