java - 线程崩溃

Question

我有一个简单的项目，我正在努力加深对 Java 中的锁定和线程的理解。本质上，我有一个缓存对象，它启动一个清理线程来删除超过给定年龄的项目。测试类“Tester”运行两个额外的线程，一个用于将项目添加到缓存，另一个用于打印缓存的内容。由于某种原因，当清理线程修改嵌入在 Cache 中的 HashMap 时，它会停止任何进一步的迭代。我已经尝试同步访问器/修改器方法以及围绕缓存中的 LOCK 对象进行同步。任何想法或帮助都将是 muy beueno。 ;)

public class Cache 
{
    private HashMap<Object, ObjectWrapper> cachedObjects = new HashMap<>();
    private static Cache cache = null;
    private static int TIME_TO_KEEP = 60000; // 60 seconds
    private final static Object LOCK = new Object();

    public static Cache getInstance()
    {
        if (cache == null)
        {
            cache = new Cache();
        }
        return cache;
    }

    private Cache()
    {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

        Runnable task = () -> {
            synchronized(LOCK)
            {
                System.out.println("Cleanup");
                Set<Object> keys = cachedObjects.keySet();
                final long now = System.currentTimeMillis();
                keys.forEach(k -> {
                    try
                    {

                        {
                            ObjectWrapper ow = cachedObjects.get(k);
                            if(ow.getExpireTime() < now)
                            {
                                int size = cachedObjects.size();
                                cachedObjects.remove(k, ow);
                                System.out.println("DEL:" + k + ", from " + size + " to " + cachedObjects.size());
                            }
                        }
                    }
                    catch (Throwable t)
                    {
                        t.printStackTrace();
                    }
                });
            }
        };

        executor.scheduleWithFixedDelay(task, 5, 15, TimeUnit.SECONDS);
    }

    public void addObject(Object key, Object obj)
    {
        synchronized(LOCK)
        {
            ObjectWrapper ow = new ObjectWrapper(obj, System.currentTimeMillis() + TIME_TO_KEEP);
            cachedObjects.put(key, ow);
        }
    }

    public ObjectWrapper getObj(Object key)
    {
        synchronized(LOCK)
        {
            return cachedObjects.get(key);
        }
    }

    public Collection<ObjectWrapper> getValues()
    {
        synchronized(LOCK)
        {
            return cachedObjects.values();
        }
    }

    public Set<Object> getKeys()
    {
        synchronized(LOCK)
        {
            return cachedObjects.keySet();
        }
    }

    public static void main(String[] args)
    {
        Cache cache = Cache.getInstance();
    }
}

import java.util.*;
import java.util.concurrent.*;

public class Tester 
{
    private static Integer id = 0;
    private static Cache cache = Cache.getInstance();

    public static void main(String[] args)
    {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

        Runnable adder = () -> {
            System.out.println("Adding id:" + ++id);
            Object o = new Object();
            cache.addObject(id, o);
        };

        executor.scheduleWithFixedDelay(adder, 5, 10, TimeUnit.SECONDS);


        Runnable tester = () -> {
            long currTime = System.currentTimeMillis();
            System.out.println("Test: ");
            Set<Object> keys = cache.getKeys();
            keys.forEach(k -> {
                ObjectWrapper o = cache.getObj(k);
                System.out.println(k + ">" + currTime + "::" + o.getExpireTime());
            });
        };

        executor.scheduleWithFixedDelay(tester, 20, 10, TimeUnit.SECONDS);
    }
}

public class ObjectWrapper
{
    private Object obj;
    private long expireTime;

    public ObjectWrapper(Object obj, long expireTime)
    {
        this.obj = obj;
        this.expireTime = expireTime;
    }

    public Object getObj()
    {
        return obj;
    }

    public void setObj(Object obj)
    {
        this.obj = obj;
    }

    public long getExpireTime()
    {
        return expireTime;
    }

    public void setExpireTime(long expireTime)
    {
        this.expireTime = expireTime;
    }
}

Answer 1

考虑使用 ConcurrentHashMap，它是一个本地线程安全的映射实现，而 HashMap 不是这种情况。

你的错误主要在这里:

public Collection<ObjectWrapper> getValues()
{
    synchronized(LOCK)
    {
        return cachedObjects.values();
    }
}

public Set<Object> getKeys()
{
    synchronized(LOCK)
    {
        return cachedObjects.keySet();
    }
}

这样做是不够的，因为您共享非线程安全集合中实际提供的 HashMap 的键和值，因此您需要在迭代它们的内容时同步访问或您可以简单地返回一个安全副本，如下所示:

public Collection<ObjectWrapper> getValues()
{
    synchronized(LOCK)
    {
        return new ArrayList<>(cachedObjects.values());
    }
}

public Set<Object> getKeys()
{
    synchronized(LOCK)
    {
        return new HashSet<>(cachedObjects.keySet());
    }
}

您还需要使 ObjectWrapper 线程安全，因为它意味着共享，否则您的代码将不再是线程安全的。最简单的方法是让它像这样不可变:

public class ObjectWrapper
{
    private final Object obj;
    private final long expireTime;

    public ObjectWrapper(Object obj, long expireTime)
    {
        this.obj = obj;
        this.expireTime = expireTime;
    }

    public Object getObj()
    {
        return obj;
    }

    public long getExpireTime()
    {
        return expireTime;
    }
}