Java并发&分布式场景

Question

我正在尝试了解我的应用程序的并发要求。现在，我正在内存中存储每个用户最近更新的地图。结构如下：

Map<String, RecentUpdates> cacheMap; //Key is userId

class RecentUpdates {
    public String userId; //the user   
    public List<EntityUpdate> recentUpdates;
}

class EntityUpdate { 
    public String timestamp; //The timestamp when the entity was updated
    public String id; //The unique entity id    
}

地图中的以下线程读/写：

单线程A：

从数据库队列（MongoDB 操作日志）中读取操作。对于每个插入/更新/删除操作：

1. 如果用户没有最近更新 缓存映射中的对象，创建它并放置 将其放入缓存映射中。
2. 创建一个新的 EntityUpdate 并添加它 到用户的最近更新列表。

单线程B：

迭代缓存映射。如果条目在过去一小时内没有最近更新，则删除该条目

多线程 C 到 Z：

如果缓存映射包含给定用户的最近更新，则迭代最近更新并检索晚于给定时间戳发生的更新。

问题是：

1.缓存映射并发

哪种数据结构最适合缓存映射？并发哈希映射？也许与番石榴不同？

2。最近更新列表并发

鉴于只有一个线程向其中添加项目，我是否需要同步最近的更新列表，或者我可以安全地使用 ArrayList 吗？

如果我是对的，如果线程 A 添加新元素，而线程 C-Z 使用 Iterator 迭代列表，则会抛出 ConcurrentModificationException。但是，如果我使用 for 循环迭代列表是否安全？

for(int i = 0; i < recentUpdates.size(); i++) {}

3.分布式场景

在分布式场景下（线程C-Z在不同的web服务器），能否根据我的需要推荐一个分布式缓存方案（Hazelcast、Terracota...）？

非常感谢

Answer 1

对于 1 和 2，您可以使用 Guava's MapMaker 来构建您的缓存：

Map<String, RecentUpdates> cacheMap = new MapMaker().expireAfterAccess(1, TimeUnit.HOURS).makeComputingMap(new Function<String, RecentUpdates>() {
  public RecentUpdates apply(String user) {
    return create(user); //whatever your impl is
  }
}

此映射保证每次获取唯一键时调用一次且仅调用一次 init 函数（如果有相同键的并发获取，则其他人阻塞并等待）。

只要通过那里完成对最近更新的所有访问并且您不保留对它们的引用，这将非常有效。

但是，您似乎想要一个短暂的（可变的）RecentUpdates，它可能会出现在此结构之外更新的问题。然后，您可以对未重置缓存中的过期时间的 RecentUpdates 结构进行更新。

解决上述问题的一种方法是在作为 ConcurrentMap 的缓存中替换不可变的 RecentUpdates

while (true) {
  RecentUpdates old = map.get(key);
  RecentUpdates updated = update(old); // copy
  if((old == null) 
      ? map.putIfAbsent(key, value) == null 
      : map.replace(key, old, value)) {
    return updated;
  }
}

这意味着地图在到期时不会有任何竞争条件。

就 3. 而言，做出这样的决定有很多考虑因素。谁还需要这个缓存？它是否仅适用于当前用户，因此您可能首先考虑使用其他类型的缓存。

Answer 2

1 ConcurrentHashMap 可以工作，但你需要使用putIfAbsent，这意味着你必须构造可能不必要的RecentUpdate 对象（如果你不小心，不必要的List 对象）。或者，使用同步的 get/put：

synchronized RecentUpdates getOrCreateRecentUpdates(String key) {
  RecentUpdates recentUpdates = map.get(key);
  if (recentUpdates == null) {
    recentUpdates = new RecentUpdates();
    map.put(key, recentUpdates);
  }
  return recentUpdates;
}

2 多个线程可以访问列表吗？如果是，那么您需要一些同步。 ArrayList 默认不同步。使用.size() 是不安全的。如果没有内存屏障（同步、易失性等），您无法保证其他线程会看到列表已更新。

我没有使用 #3 的经验。

Answer 3

我尝试了一个解决方案。这比使用锁的解决方案使用更多的内存，但提供更好的并发性。请看看这是否符合您的需求。

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Stack;
import java.util.concurrent.ConcurrentHashMap;

public class CacheMap {
    public static class EntityUpdate {
        public final String entityId;
        public final long timestamp;
        public final EntityUpdate previous;

        public EntityUpdate(String entityId, 
                            long timestamp, EntityUpdate previous) {
            this.entityId = entityId;
            this.timestamp = timestamp;
            this.previous = previous;
        }

        public EntityUpdate cloneChangingPrevious(EntityUpdate newPrevious) {
            return new EntityUpdate(entityId, timestamp, newPrevious);
        }
    }

    public static class RecentUpdates {
        public final String userId;
        public final EntityUpdate lastUpdate;

        public RecentUpdates(String userId, EntityUpdate lastUpdate) {
            this.userId = userId;
            this.lastUpdate = lastUpdate;
        }

        public RecentUpdates recordUpdate(String entityId, long timestamp) {
            EntityUpdate update = new EntityUpdate(entityId, 
                timestamp, lastUpdate);
            return new RecentUpdates(userId, update);
        }

        public RecentUpdates removeUpdatesOlderThan(long timestamp) {
            Stack<EntityUpdate> recent = new Stack<EntityUpdate>();
            EntityUpdate update = lastUpdate;
            while (update != null) {
                if (update.timestamp >= timestamp) {
                    recent.push(update);
                } else {
                    break;
                }
                update = update.previous;
            }

            EntityUpdate last = null;
            while (!recent.isEmpty()) {
                last = recent.pop().cloneChangingPrevious(last);
            }

            return new RecentUpdates(userId, last);
        }

        public boolean isEmpty() {
            return lastUpdate == null;
        }

        public List<EntityUpdate> getUpdatesSince(long timestamp) {
            List<EntityUpdate> list = new ArrayList<EntityUpdate>();
            EntityUpdate update = lastUpdate;
            while (update != null) {
                if (update.timestamp >= timestamp) {
                    list.add(update);
                } else {
                    break;
                }
                update = update.previous;
            }
            return Collections.unmodifiableList(list);
        }
    }

    private final ConcurrentHashMap<String, RecentUpdates> map = 
        new ConcurrentHashMap<String, RecentUpdates>();

    // called by thread A
    public void recordUpdate(String userId, String entityId) {
        boolean done = false;
        while (!done) {
            RecentUpdates updates = map.get(userId);
            if (updates == null) {
                // looks like there is no mapping for this userId,
                // make an effort to insert a new one
                map.putIfAbsent(userId, new RecentUpdates(userId, null));
            }
            // query the map again
            updates = map.get(userId);
            // updates could still be null, because the entry might have
            // been removed from the map by now; if so, retry
            if (updates != null) {
                long newTimestamp = System.currentTimeMillis();
                RecentUpdates newVal = 
                    updates.recordUpdate(entityId, newTimestamp);
                done = map.replace(userId, updates, newVal);
            }
        }
    }

    // called by thread B
    public void removeUpdatesOlderThan(long timestamp) {
        for (String userId : map.keySet()) {
            boolean done = false;
            while (!done) {
                // updates will always be non-null, 
                // because only this thread can
                // remove an entry from the map
                RecentUpdates updates = map.get(userId);
                RecentUpdates purgedUpdates = 
                    updates.removeUpdatesOlderThan(timestamp);
                if (purgedUpdates.isEmpty()) {
                    // remove from the map, if now new insert has
                    // happened in the interim
                    done = map.remove(userId, updates);
                } else {
                    // replace with the purged value, if no new
                    // insert has happened in the interim
                    done = map.replace(userId, updates, purgedUpdates);
                }
            }
        }
    }

    // called by threads C-Z
    public List<EntityUpdate> getUpdatesSince(String userId, long timestamp) {
        RecentUpdates updates = map.get(userId);
        if (updates == null) {
            return Collections.EMPTY_LIST;
        } else {
            return updates.getUpdatesSince(timestamp);
        }

    }
}