[ConcurrentHashMap]：地图更新/获取操作在单线程操作中从地图返回陈旧值

Question

对于某些key，地图仍然具有陈旧价值。一些键没有反映新的更新。此错误属于not reproducible 类别。

代码：

 class DemoCache{

    private ConcurrentHashMap<String,Demo> demoByName = new ConcurrentHashMap<String,Demo>();
    private ConcurrentHashMap<String,Demo> demoByID = new ConcurrentHashMap<String,Demo>();

    public initializeFromDB(){

        log.info(me + "Initializing/refresh  instrument from database.");
        DemoDbDynamo demoDbDynamo = new DemoDbDynamo();

        final AtomicInteger progressCounter = new AtomicInteger();
        try
        {       
            demoDbDynamo.listAll()
                .stream()
                .peek(i -> progressCounter.incrementAndGet())
                .forEach(this::updateCache);            
        }
        catch (Exception e)
        {
            log.error(me + "Exception fetching demo cache  from table. " + e );
        }
        log.info(me + "count: " + progressCounter.get());

      }

    }

    public void updateCache(Demo demo){

        Demo existing = demoByID.get(demo.getID());

        demoByID.put( dmeo.getID(), demo );
        // this is an updated existing demo
        if(existing != null) {
            //if name have changed
            demoByName.remove(existing.getName());
        }

        demoByName.put      ( demo.getName(),       demo );
        //logging new value -- demo.getName and demo.getValue
        //logging old value - demo.getName and demo.getValue

    }
}

地图大小：6k

整个操作是单线程的。 JMS 主题命中此类以从 DB 初始化地图。 我们有 4 个不同的服务器，每个服务器都有这个本地缓存，由 JMS 主题消息刷新。在 4 个服务器中，有 3 个服务器使用所有正确的值进行更新，而 1 个服务器仍然保留少数键的陈旧值。

这个问题的根本原因是什么？

评论结果更新：

日志：

JMS msg

ip-10-0-33-185 20:15:40.374 [ThreadName=ActiveMQ Session Task-72] DEBUG cache.DemoCache {} -- DemoCache.onMessage() : [msg=<response mt='5099'/>]

ip-10-0-33-185 20:15:40.375 [ThreadName=ActiveMQ Session Task-72] INFO  cache.DemoCache {} -- DemoCache.initializeFromDB(): Initializing/refresh  cache from database.

ip-10-0-33-185 20:15:45.897 [ThreadName=ActiveMQ Session Task-72] DEBUG cache.DemoCache {} -- 
[newObject=[NewID=06926627-e950-48f3-9c53-b679f61120ec newName=foo,newValue=2640.98]]
[OldObject=[oldName=foo,oldValue=2641.05]]

ip-10-0-33-185 20:15:45.913 [ActiveMQ Session Task-72] INFO  cache.DemoCache {} -- DemoCache.initializeFromDB(): count: 5362

这里 _collector = ip-10-0-33-185 即服务器节点。 此服务器返回旧值 = 2641.05 而不是新值。 只有一个线程在运行ActiveMQ Session Task-72。我没有看到任何其他线程。

Answer 1

尽管使用了 ConcurrentHashMap，但代码不是线程安全的。

这是一种可能发生不一致的情况。假设缓存包含 ID = "X" 的值 demo0。假设有 2 个线程，A 和 B。线程 A 具有具有相同 ID 的下一个版本的 ibject，例如 demo1。线程 B 知道这个demo1，同时获得了它的更新版本，demo2。

现在两个线程都想更新缓存。

线程 A 使用值 demo1 调用 updateCache()。在它调用Demo existing = ... 之后，在它调用demoByID.put(...) 之前，这个线程被挂起，线程B 被执行。线程 B 调用 updateCache() 值 demo2，不会被中断并成功将值 demo2 放入缓存。

现在线程 A 继续。它执行demoByID.put(...) 等。但它适用于demo1。因此它将demo1 放入缓存，从而替换demoByID 和demoByName 中的新版本demo2。

你能做什么？

检查存在和修改缓存之间的所有操作都应该作为一个带有单个线程的块来完成。例如，对方法updateCache() 的全部内容使用lock 或使用声明方法updateCache() 同步。