我有一個場景，我必須維護一個Map可以由多個線程填充的，每個線程都修改它們各自的List（唯一標識符/鍵是線程名稱），并且當線程的列表大小超過固定的批量大小時，我們必須將記錄持久化到數(shù)據(jù)庫中。聚合器類private volatile ConcurrentHashMap<String, List<T>> instrumentMap = new ConcurrentHashMap<String, List<T>>();private ReentrantLock lock ;public void addAll(List<T> entityList, String threadName) { try { lock.lock(); List<T> instrumentList = instrumentMap.get(threadName); if(instrumentList == null) { instrumentList = new ArrayList<T>(batchSize); instrumentMap.put(threadName, instrumentList); } if(instrumentList.size() >= batchSize -1){ instrumentList.addAll(entityList); recordSaver.persist(instrumentList); instrumentList.clear(); } else { instrumentList.addAll(entityList); } } finally { lock.unlock(); }}每 2 分鐘后運行一個單獨的線程（使用相同的鎖）來持久化所有記錄Map（以確保每 2 分鐘后有一些東西持久化并且地圖大小不會變得太大）if(//Some condition) { Thread.sleep(//2 minutes); aggregator.getLock().lock(); List<T> instrumentList = instrumentMap.values().stream().flatMap(x->x.stream()).collect(Collectors.toList()); if(instrumentList.size() > 0) { saver.persist(instrumentList); instrumentMap .values().parallelStream().forEach(x -> x.clear()); aggregator.getLock().unlock(); }}這個解決方案在我們測試的幾乎所有場景中都可以正常工作，除了有時我們會看到一些記錄丟失了，即它們根本沒有持久化，盡管它們被很好地添加到了地圖中。我的問題是：這段代碼有什么問題？這里不是ConcurrentHashMap最好的解決方案嗎？List與 一起使用的那個ConcurrentHashMap有問題嗎？我應該使用這里的計算方法嗎ConcurrentHashMap（我認為不需要，因為ReentrantLock已經(jīng)在做同樣的工作了）？
3 回答
慕運維8079593
TA貢獻1876條經(jīng)驗 獲得超5個贊
我們讓instrumentList實例以非同步方式逃逸，即訪問/操作在列表上發(fā)生而沒有任何同步。通過將副本傳遞給其他方法來修復相同的問題就可以了。
以下代碼行是發(fā)生此問題的代碼行
recordSaver.persist(instrumentList); 儀器列表.clear();
在這里，我們允許instrumentList實例以非同步方式逃逸，即它被傳遞到另一個類（recordSaver.persist），在該類中對其進行操作，但我們也在下一行（在 Aggregator 類中）清除列表，并且所有這些都以非同步方式發(fā)生。無法在記錄保護程序中預測列表狀態(tài)......一個非常愚蠢的錯誤。
我們通過將instrumentList的克隆副本傳遞給 recordSaver.persist(...) 方法來解決此問題。這樣，instrumentList.clear()不會影響 recordSaver 中可用的列表以進行進一步操作。
尚方寶劍之說
TA貢獻1788條經(jīng)驗 獲得超4個贊
看起來這是對不需要的優(yōu)化的嘗試。在這種情況下，越少越好，越簡單越好。在下面的代碼中，僅使用了兩個并發(fā)概念：synchronized確保正確更新共享列表并final確保所有線程看到相同的值。
import java.util.ArrayList;
import java.util.List;
public class Aggregator<T> implements Runnable {
private final List<T> instruments = new ArrayList<>();
private final RecordSaver recordSaver;
private final int batchSize;
public Aggregator(RecordSaver recordSaver, int batchSize) {
super();
this.recordSaver = recordSaver;
this.batchSize = batchSize;
}
public synchronized void addAll(List<T> moreInstruments) {
instruments.addAll(moreInstruments);
if (instruments.size() >= batchSize) {
storeInstruments();
}
}
public synchronized void storeInstruments() {
if (instruments.size() > 0) {
// in case recordSaver works async
// recordSaver.persist(new ArrayList<T>(instruments));
// else just:
recordSaver.persist(instruments);
instruments.clear();
}
}
@Override
public void run() {
while (true) {
try { Thread.sleep(1L); } catch (Exception ignored) {
break;
}
storeInstruments();
}
}
class RecordSaver {
void persist(List<?> l) {}
}
}
慕虎7371278
TA貢獻1802條經(jīng)驗 獲得超4個贊
我明白了，您parallelStream在鎖中使用了 ConcurrentHashMap。我不了解 Java 8+ 流支持，但快速搜索顯示，
ConcurrentHashMap 是一種復雜的數(shù)據(jù)結(jié)構(gòu)，過去曾經(jīng)存在并發(fā)錯誤
并行流必須遵守復雜且記錄不充分的使用限制
您正在并行流中修改數(shù)據(jù)
基于這些信息（以及我的直覺驅(qū)動的并發(fā)錯誤檢測器?），我敢打賭，刪除對的調(diào)用parallelStream可能會提高代碼的健壯性。此外，正如@Slaw 所提到的，如果所有instrumentMap使用都已被鎖保護，則應該使用普通 HashMap 代替 ConcurrentHashMap 。
當然，由于您沒有發(fā)布 的代碼recordSaver，因此它也有可能存在錯誤（不一定是與并發(fā)相關(guān)的錯誤）。特別是，您應該確保從持久存儲中讀取記錄的代碼（用于檢測記錄丟失的代碼）是安全、正確的，并且與系統(tǒng)的其余部分正確同步（最好使用健壯的, 行業(yè)標準的 SQL 數(shù)據(jù)庫）。
添加回答
舉報
0/150
提交
取消