c# - 引发事件时我应该如何实现 "quiet period"？

Question

我正在使用订阅者/通知者模式在 C# 中引发和使用来 self 的 .Net 中间层的事件。一些事件以“突发”方式引发，例如，当数据从导入文件的批处理程序中持久化时。这会执行一个可能长时间运行的任务，我想通过实现“静默期”来避免每秒多次触发事件，事件系统会一直等到事件流变慢才能处理事件。

当发布者主动通知订阅者时，我应该怎么做？我不想等到有事件进来才去查看是否还有其他人在等待静默期...

目前没有主机进程来轮询订阅模型。我应该放弃发布/订阅模式还是有更好的方法？

Answer 1

这是一个粗略的实现，可能会为您指明方向。在我的示例中，涉及通知的任务是保存数据对象。保存对象时，将引发 Saved 事件。除了简单的 Save 方法之外，我还实现了 BeginSave 和 EndSave 方法以及与这两个方法一起用于批量保存的 Save 重载。调用 EndSave 时，将触发单个 BatchSaved 事件。

显然，您可以更改它以满足您的需要。在我的示例中，我跟踪了在批处理操作期间保存的所有对象的列表，但这可能不是您需要做的事情......您可能只关心保存了多少对象，甚至只是关心批量保存操作已完成。如果您预期要保存大量对象，那么将它们存储在我的示例中的列表中可能会成为内存问题。

编辑:我在示例中添加了一个“阈值”概念，试图防止内存中保存大量对象。但是，这会导致 BatchSaved 事件更频繁地触发。我还添加了一些锁定来解决潜在的线程安全问题，尽管我可能在那里遗漏了一些东西。

class DataConcierge<T>
{
    // *************************
    // Simple save functionality
    // *************************

    public void Save(T dataObject)
    {
        // perform save logic

        this.OnSaved(dataObject);
    }

    public event DataObjectSaved<T> Saved;

    protected void OnSaved(T dataObject)
    {
        var saved = this.Saved;
        if (saved != null)
            saved(this, new DataObjectEventArgs<T>(dataObject));
    }

    // ************************
    // Batch save functionality
    // ************************

    Dictionary<BatchToken, List<T>> _BatchSavedDataObjects = new Dictionary<BatchToken, List<T>>();
    System.Threading.ReaderWriterLockSlim _BatchSavedDataObjectsLock = new System.Threading.ReaderWriterLockSlim();

    int _SavedObjectThreshold = 17; // if the number of objects being stored for a batch reaches this threshold, then those objects are to be cleared from the list.

    public BatchToken BeginSave()
    {
        // create a batch token to represent this batch
        BatchToken token = new BatchToken();

        _BatchSavedDataObjectsLock.EnterWriteLock();
        try
        {
            _BatchSavedDataObjects.Add(token, new List<T>());
        }
        finally
        {
            _BatchSavedDataObjectsLock.ExitWriteLock();
        }
        return token;
    }

    public void EndSave(BatchToken token)
    {
        List<T> batchSavedDataObjects;
        _BatchSavedDataObjectsLock.EnterWriteLock();
        try
        {
            if (!_BatchSavedDataObjects.TryGetValue(token, out batchSavedDataObjects))
                throw new ArgumentException("The BatchToken is expired or invalid.", "token");

            this.OnBatchSaved(batchSavedDataObjects); // this causes a single BatchSaved event to be fired

            if (!_BatchSavedDataObjects.Remove(token))
                throw new ArgumentException("The BatchToken is expired or invalid.", "token");
        }
        finally
        {
            _BatchSavedDataObjectsLock.ExitWriteLock();
        }
    }

    public void Save(BatchToken token, T dataObject)
    {
        List<T> batchSavedDataObjects;
        // the read lock prevents EndSave from executing before this Save method has a chance to finish executing
        _BatchSavedDataObjectsLock.EnterReadLock();
        try
        {
            if (!_BatchSavedDataObjects.TryGetValue(token, out batchSavedDataObjects))
                throw new ArgumentException("The BatchToken is expired or invalid.", "token");

            // perform save logic

            this.OnBatchSaved(batchSavedDataObjects, dataObject);
        }
        finally
        {
            _BatchSavedDataObjectsLock.ExitReadLock();
        }
    }

    public event BatchDataObjectSaved<T> BatchSaved;

    protected void OnBatchSaved(List<T> batchSavedDataObjects)
    {
        lock (batchSavedDataObjects)
        {
            var batchSaved = this.BatchSaved;
            if (batchSaved != null)
                batchSaved(this, new BatchDataObjectEventArgs<T>(batchSavedDataObjects));
        }
    }

    protected void OnBatchSaved(List<T> batchSavedDataObjects, T savedDataObject)
    {
        // add the data object to the list storing the data objects that have been saved for this batch
        lock (batchSavedDataObjects)
        {
            batchSavedDataObjects.Add(savedDataObject);

            // if the threshold has been reached
            if (_SavedObjectThreshold > 0 && batchSavedDataObjects.Count >= _SavedObjectThreshold)
            {
                // then raise the BatchSaved event with the data objects that we currently have
                var batchSaved = this.BatchSaved;
                if (batchSaved != null)
                    batchSaved(this, new BatchDataObjectEventArgs<T>(batchSavedDataObjects.ToArray()));

                // and clear the list to ensure that we are not holding on to the data objects unnecessarily
                batchSavedDataObjects.Clear();
            }
        }
    }
}

class BatchToken
{
    static int _LastId = 0;
    static object _IdLock = new object();

    static int GetNextId()
    {
        lock (_IdLock)
        {
            return ++_LastId;
        }
    }

    public BatchToken()
    {
        this.Id = GetNextId();
    }

    public int Id { get; private set; }
}

class DataObjectEventArgs<T> : EventArgs
{
    public T DataObject { get; private set; }

    public DataObjectEventArgs(T dataObject)
    {
        this.DataObject = dataObject;
    }
}

delegate void DataObjectSaved<T>(object sender, DataObjectEventArgs<T> e);

class BatchDataObjectEventArgs<T> : EventArgs
{
    public IEnumerable<T> DataObjects { get; private set; }

    public BatchDataObjectEventArgs(IEnumerable<T> dataObjects)
    {
        this.DataObjects = dataObjects;
    }
}

delegate void BatchDataObjectSaved<T>(object sender, BatchDataObjectEventArgs<T> e);

在我的示例中，我选择使用 token 概念来创建单独的批处理。这允许在单独线程上运行的较小批量操作完成并引发事件，而无需等待较大批量操作完成。

我制作了单独的事件:Saved 和 BatchSaved。但是，这些可以很容易地合并到一个事件中。

编辑:Steven Sudit 在访问事件代表时指出的固定竞争条件。

编辑:修改我示例中的锁定代码以使用 ReaderWriterLockSlim 而不是 Monitor(即“锁定”语句)。我认为存在一些竞争条件，例如在 Save 和 EndSave 方法之间。 EndSave 可能会执行，导致数据对象列表从字典中删除。如果 Save 方法同时在另一个线程上执行，则有可能将数据对象添加到该列表，即使它已从字典中删除。

在我修改过的例子中，这种情况是不会发生的，如果在EndSave之后执行Save方法，就会抛出异常。这些竞争条件主要是由于我试图避免我认为不必要的锁定造成的。我意识到更多代码需要放在一个锁中，但决定使用 ReaderWriterLockSlim 而不是 Monitor，因为我只想防止 Save 和 EndSave 同时执行；不需要阻止多个线程同时执行 Save。请注意，Monitor 仍用于同步访问从字典中检索到的特定数据对象列表。

编辑:添加用法示例

下面是上述示例代码的使用示例。

    static void DataConcierge_Saved(object sender, DataObjectEventArgs<Program.Customer> e)
    {
        Console.WriteLine("DataConcierge<Customer>.Saved");
    }

    static void DataConcierge_BatchSaved(object sender, BatchDataObjectEventArgs<Program.Customer> e)
    {
        Console.WriteLine("DataConcierge<Customer>.BatchSaved: {0}", e.DataObjects.Count());
    }

    static void Main(string[] args)
    {
        DataConcierge<Customer> dc = new DataConcierge<Customer>();
        dc.Saved += new DataObjectSaved<Customer>(DataConcierge_Saved);
        dc.BatchSaved += new BatchDataObjectSaved<Customer>(DataConcierge_BatchSaved);

        var token = dc.BeginSave();
        try
        {
            for (int i = 0; i < 100; i++)
            {
                var c = new Customer();
                // ...
                dc.Save(token, c);
            }
        }
        finally
        {
            dc.EndSave(token);
        }
    }

这导致了以下输出:

DataConcierge<Customer>.BatchSaved: 17

DataConcierge<Customer>.BatchSaved: 17

DataConcierge<Customer>.BatchSaved: 17

DataConcierge<Customer>.BatchSaved: 17

DataConcierge<Customer>.BatchSaved: 17

DataConcierge<Customer>.BatchSaved: 15

我的示例中的阈值设置为 17，因此一批 100 个项目会导致 BatchSaved 事件触发 6 次。