我正在使用订阅者/通知程序模式来提升和使用C#中的.Net中间层事件.例如,当从导入文件的批处理程序中持久保存数据时,某些事件会以"突发"形式引发.这执行一个可能长时间运行的任务,我想通过实现"静默期"来避免每秒多次触发事件,从而事件系统等待直到事件流减慢以处理事件.
当发布者在通知订阅者方面发挥积极作用时,我该如何做？我不想等到事件进来检查是否还有其他人等待安静的时期...
目前没有主机进程来轮询订阅模型.我应该放弃发布/订阅模式还是有更好的方法？
这是一个粗略的实现，可能会为您指明方向。在我的示例中，涉及通知的任务是保存数据对象。保存对象时，将引发 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);
在我的示例中，我选择使用令牌概念来创建单独的批次。这允许在单独的线程上运行的较小批处理操作完成并引发事件，而无需等待较大批处理操作完成。
我做了单独的事件：Saved 和 BatchSaved。然而，这些可以很容易地合并到一个事件中。
编辑：修复了 Steven Sudit 在访问事件代表时指出的竞争条件。
编辑：在我的示例中修改锁定代码以使用 ReaderWriterLockSlim 而不是 Monitor （即“lock”语句）。我认为存在一些竞争条件，例如 Save 和 EndSave 方法之间。EndSave 的执行可能会导致数据对象列表从字典中删除。如果 Save 方法同时在另一个线程上执行，则数据对象可能会被添加到该列表中，即使它已从字典中删除。
在我修改后的示例中，这种情况不会发生，并且如果 Save 方法在 EndSave 之后执行，则会抛出异常。这些竞争条件主要是由于我试图避免我认为不必要的锁定而引起的。我意识到锁内需要更多代码，但决定使用 ReaderWriterLockSlim 而不是 Monitor，因为我只想防止 Save 和 EndSave 同时执行；不需要阻止多个线程同时执行 Save。请注意，监视器仍然用于同步对从字典中检索的数据对象的特定列表的访问。
编辑：添加使用示例
下面是上述示例代码的使用示例。
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<客户>.BatchSaved：17
DataConcierge<客户>.BatchSaved：17
DataConcierge<客户>.BatchSaved：17
DataConcierge<客户>.BatchSaved：17
DataConcierge<客户>.BatchSaved：17
DataConcierge<客户>.BatchSaved：15
我的示例中的阈值设置为 17，因此一批 100 个项目会导致 BatchSaved 事件触发 6 次。