并发字典的线程安全重新初始化

Question

我想知道以下代码是否是线程安全的，我认为不是。我怎么可能让它线程安全？

基本上我有一个ConcurrentDictionary，它充当数据库表的缓存。我想每 10 秒查询一次数据库并更新数据库缓存。将有其他线程一直在查询该字典。

我不能只使用TryAdd，因为可能还有我已删除的元素。所以我决定不要搜索整个字典来更新、添加或删除。我只是重新初始化字典。请告诉我这是不是一个愚蠢的想法。

我担心的是，当我重新初始化字典时，当初始化发生时，查询线程将不再是线程安全的实例。出于这个原因，我在更新字典时使用了一个锁，但是我不确定这是否正确，因为锁中的对象发生了变化？

private static System.Timers.Timer updateTimer;
private static volatile Boolean _isBusyUpdating = false;
private static ConcurrentDictionary<int, string> _contactIdNames;

public Constructor()
{
    // Setup Timers for data updater         
    updateTimer = new System.Timers.Timer();
    updateTimer.Interval = new TimeSpan(0, 0, 10, 0).TotalMilliseconds;
    updateTimer.Elapsed += OnTimedEvent;
    // Start the timer
    updateTimer.Enabled = true;
}

private void OnTimedEvent(Object source, System.Timers.ElapsedEventArgs e)
{
    if (!_isBusyUpdating)
    {
        _isBusyUpdating = true;
        // Get new data values and update the list
        try
        {
            var tmp = new ConcurrentDictionary<int, string>();
            using (var db = new DBEntities())
            {
                foreach (var item in db.ContactIDs.Select(x => new { x.Qualifier, x.AlarmCode, x.Description }).AsEnumerable())
                {
                    int key = (item.Qualifier * 1000) + item.AlarmCode;
                    tmp.TryAdd(key, item.Description);
                }
            }
            if (_contactIdNames == null)
            {
                _contactIdNames = tmp;
            }
            else
            {
                lock (_contactIdNames)
                {
                    _contactIdNames = tmp;
                }
            }
        }
        catch (Exception e)
        {
            Debug.WriteLine("Error occurred in update ContactId db store", e);
        }
        _isBusyUpdating = false;
    }
}

    /// Use the dictionary from another Thread
    public int GetIdFromClientString(string Name)
    {
        try
        {
            int pk;
            if (_contactIdNames.TryGetValue(Name, out pk))
            {
                return pk;
            }
        }
        catch { }
        //If all else fails return -1
        return -1;
    }

Answer 1

你是对的，你的代码不是线程安全的。

1. 您需要锁定_isBusyUpdating 变量。
2. 您每次都需要锁定_contactIdNames，而不仅仅是当它不是null时。

此外，这段代码与singleton pattern 类似，并且在初始化时也存在同样的问题。您可以使用Double checked locking 解决它。但是，您在访问条目时还需要双重检查锁定。

在一次更新整个字典的情况下，每次访问时都需要锁定当前值。否则，您可以在它仍在更改并出现错误时访问它。所以你要么每次都需要锁定变量，要么使用Interlocked。

因为MSDN says volatile 应该使用_isBusyUpdating 来解决问题，所以它应该是线程安全的。

如果您不想跟踪_contactIdNames 线程安全，请尝试对同一字典中的每个条目进行更新。问题将在于数据库和当前值之间的差异检测（哪些条目已被删除或添加，其他可以简单重写），但不在于线程安全，因为ConcurrentDictionary 已经是线程安全的。

Answer 2

您似乎为自己做了很多工作。以下是我将如何处理这项任务：

public class Constructor
{
    private volatile Dictionary<int, string> _contactIdNames;

    public Constructor()
    {
        Observable
            .Interval(TimeSpan.FromSeconds(10.0))
            .StartWith(-1)
            .Select(n =>
            {
                using (var db = new DBEntities())
                {
                    return db.ContactIDs.ToDictionary(
                        x => x.Qualifier * 1000 + x.AlarmCode,
                        x => x.Description);
                }
            })
            .Subscribe(x => _contactIdNames = x);
    }

    public string TryGetValue(int key)
    {
        string value = null;
        _contactIdNames.TryGetValue(key, out value);
        return value;
    }
}

我正在使用 Microsoft 的响应式扩展 (Rx) 框架 - NuGet "Rx-Main" - 用于更新字典的计时器。

Rx 应该相当简单。如果您以前从未见过它，简单来说就像 LINQ 遇到事件一样。

如果您不喜欢 Rx，那么就使用您当前的计时器模型。

这段代码所做的只是每 10 秒从数据库中创建一个新字典。我只是使用一个普通的字典，因为它只是从一个线程创建的。由于引用分配是原子的，因此您可以根据需要在完全线程安全的情况下重新分配字典。

只要元素不变，多个线程就可以安全地从字典中读取。

Answer 3

我想知道下面的代码是否是线程安全的，我假设它是线程安全的 不是。我怎么可能让它线程安全？

我相信不是。首先，我会为ConcurrentDictionary 创建属性并检查get 方法中是否正在进行更新，如果是，我会返回之前版本的对象：

    private object obj = new object();
    private ConcurrentDictionary<int, string> _contactIdNames;
    private ConcurrentDictionary<int, string> _contactIdNamesOld;
    private volatile bool _isBusyUpdating = false;

    public ConcurrentDictionary<int, string> ContactIdNames
    {
        get
        {
            if (!_isBusyUpdating) return _contactIdNames;

            return _contactIdNamesOld;
        }
        private set 
        {
            if(_isBusyUpdating) _contactIdNamesOld = 
                new ConcurrentDictionary<int, string>(_contactIdNames);

            _contactIdNames = value; 
        }
    }

你的方法可以是：

    private static void OnTimedEvent(Object source, System.Timers.ElapsedEventArgs e)
    {
        if (_isBusyUpdating) return;

        lock (obj)
        {

            _isBusyUpdating = true;
            // Get new data values and update the list

            try
            {
                ContactIdNames = new ConcurrentDictionary<int, string>();
                using (var db = new DBEntities())
                {
                    foreach (var item in db.ContactIDs.Select(x => new { x.Qualifier, x.AlarmCode, x.Description }).AsEnumerable())
                    {
                        int key = (item.Qualifier * 1000) + item.AlarmCode;
                        _contactIdNames.TryAdd(key, item.Description);
                    }
                }                    
            }
            catch (Exception e)
            {
                Debug.WriteLine("Error occurred in update ContactId db store", e);        
                _contactIdNames = _contactIdNamesOld;            
            }
            finally 
            {                   
                _isBusyUpdating = false;
            }
        }
    }

附言

我担心的是，当我重新初始化字典时，查询 线程将不再通过线程安全的实例当 初始化发生。出于这个原因，我使用了锁 更新字典时，但是我不确定这是否正确 随着对象在锁中的变化？

ConcurrentDictionary<T> 类型是线程安全的，而不是它的实例，因此即使您创建新实例并更改对它的引用 - 也不必担心。