[.NET领域驱动设计实战系列]专题八：DDD案例：网上书店分布式消息队列和分布式缓存的实现

public class MsmqEventBus : DisposableObject, IEventBus
{
public void Publish<TMessage>(TMessage message) where TMessage : class, IEvent
{
// 将消息放入Message中Body属性进行序列化发送到消息队列中
var msmqMessage = new Message(message) { Formatter = new XmlMessageFormatter(new[] { message.GetType() }), Label = message.GetType().ToString()};
_messageQueue.Send(msmqMessage);
_committed = false;
} public void Publish<TMessage>(IEnumerable<TMessage> messages) where TMessage : class, IEvent
{
messages.ToList().ForEach(m =>
{
_messageQueue.Send(m);
_committed = false;
});
} public void Commit()
{
if (this._useInternalTransaction)
{
using (var transaction = new MessageQueueTransaction())
{
try
{
transaction.Begin();
var message = _messageQueue.Receive();
if (message != null)
{
message.Formatter = new XmlMessageFormatter(new[] { typeof(string) });
var evntType = ConvertStringToType(message.Body.ToString());
var method = _publishMethod.MakeGenericMethod(evntType);
var evnt = Activator.CreateInstance(evntType);
method.Invoke(_aggregator, new object[] { evnt }); transaction.Commit();
}
}
catch
{
transaction.Abort();
throw;
}
}
}
else
{
// 从msmq消息队列中出队，此时获得的对象是消息对象
var message = _messageQueue.Receive();
if (message != null)
{
// 指定反序列化的对象，由于我们之前把对应的事件类型保存在MessageQueue中的Label属性
// 所以此时可以通过Label属性来获得目标序列化类型
message.Formatter = new XmlMessageFormatter(new[] { ConvertStringToType(message.Label) }); // 这样message.Body获得就是对应的事件对象，后面的处理逻辑就和EventBus一样了
var evntType =message.Body.GetType();
var method = _publishMethod.MakeGenericMethod(evntType);
method.Invoke(_aggregator, new object[] { message.Body });
}
} _committed = true;
}
}

<!--Event Bus-->
<!--<register type="OnlineStore.Events.Bus.IEventBus, OnlineStore.Events" mapTo="OnlineStore.Events.Bus.EventBus, OnlineStore.Events">
<lifetime type="singleton" />
</register>--> <!--注入MsmqEventBus-->
<register type="OnlineStore.Events.Bus.IEventBus, OnlineStore.Events"
mapTo="OnlineStore.Events.Bus.MsmqEventBus, OnlineStore.Events">
<lifetime type="singleton" />
<constructor>
<param name="path" value=".\Private$\OnlineStoreQueue" />
</constructor>
</register>
</container>

// 缓存接口的定义
public interface ICacheProvider
{
/// <summary>
/// 向缓存中添加一个对象
/// </summary>
/// <param name="key">缓存的键值</param>
/// <param name="valueKey">缓存值的键值</param>
/// <param name="value">缓存的对象</param>
void Add(string key, string valueKey, object value);
void Update(string key, string valueKey, object value);
object Get(string key, string valueKey);
void Remove(string key);
bool Exists(string key);
bool Exists(string key, string valueKey);
}

// 表示基于Microsoft Patterns & Practices - Enterprise Library Caching Application Block的缓存机制的实现
// 该类简单理解为对Enterprise Library Caching中的CacheManager封装
// 该缓存实现不支持分布式缓存，更多信息参考:
// http://stackoverflow.com/questions/7799664/enterpriselibrary-caching-in-load-balance
public class EntLibCacheProvider : ICacheProvider
{
// 获得CacheManager实例，该实例的注册通过cachingConfiguration进行注册进去的，具体看配置文件
private readonly ICacheManager _cacheManager = CacheFactory.GetCacheManager(); #region ICahceProvider public void Add(string key, string valueKey, object value)
{
Dictionary<string, object> dict = null;
if (_cacheManager.Contains(key))
{
dict = (Dictionary<string, object>) _cacheManager[key];
dict[valueKey] = value;
}
else
{
dict = new Dictionary<string, object> { { valueKey, value }};
} _cacheManager.Add(key, dict);
} public void Update(string key, string valueKey, object value)
{
Add(key, valueKey, value);
} public object Get(string key, string valueKey)
{
if (!_cacheManager.Contains(key)) return null;
var dict = (Dictionary<string, object>)_cacheManager[key];
if (dict != null && dict.ContainsKey(valueKey))
return dict[valueKey];
else
return null;
} // 从缓存中移除对象
public void Remove(string key)
{
_cacheManager.Remove(key);
} // 判断指定的键值的缓存是否存在
public bool Exists(string key)
{
return _cacheManager.Contains(key);
} // 判断指定的键值和缓存键值的缓存是否存在
public bool Exists(string key, string valueKey)
{
return _cacheManager.Contains(key) &&
((Dictionary<string, object>)_cacheManager[key]).ContainsKey(valueKey);
}
#endregion
}

// 分布式缓存，该类是对微软分布式缓存服务的封装
// 在该案例中没用用到该缓存，但是提供在这里让大家明白微软的分布式缓存实现，并不是只有memcached和Redis
// Redis参考：http://www.cnblogs.com/ceecy/p/3279407.html 和 http://blog.csdn.net/suifeng3051/article/details/23739295
// 关于微软分布式缓存更多介绍参考：http://www.cnblogs.com/shanyou/archive/2010/06/29/AppFabricCaching.html
// 和http://www.cnblogs.com/mlj322/archive/2010/04/05/1704624.html
public class AppfabricCacheProvider : ICacheProvider
{
private readonly DataCacheFactory _factory = new DataCacheFactory();
private readonly DataCache _cache; public AppfabricCacheProvider()
{
this._cache = _factory.GetDefaultCache();
} #region ICacheProvider Members
public void Add(string key, string valueKey, object value)
{
// DataCache中不包含Contain方法，所有用Get方法来判断对应的key值是否在缓存中存在
var val = (Dictionary<string, object>)_cache.Get(key);
if (val == null)
{
val = new Dictionary<string, object> {{ valueKey, value}};
_cache.Add(key, val);
}
else
{
if (!val.ContainsKey(valueKey))
val.Add(valueKey, value);
else
val[valueKey] = value; _cache.Put(key, val);
}
} public void Update(string key, string valueKey, object value)
{
Add(key, valueKey, value);
} public object Get(string key, string valueKey)
{
return Exists(key, valueKey) ? ((Dictionary<string, object>)_cache.Get(key))[valueKey] : null;
} public void Remove(string key)
{
_cache.Remove(key);
} public bool Exists(string key)
{
return _cache.Get(key) != null;
} public bool Exists(string key, string valueKey)
{
var val = _cache.Get(key);
if (val == null)
return false;
return ((Dictionary<string, object>)val).ContainsKey(valueKey);
} #endregion
}

<unity xmlns="http://schemas.microsoft.com/practices/2010/unity">
<sectionExtension type="Microsoft.Practices.Unity.InterceptionExtension.Configuration.InterceptionConfigurationExtension, Microsoft.Practices.Unity.Interception.Configuration" />
<container>
<extension type="Interception" /> <!--Cache Provider-->
<register type="OnlineStore.Infrastructure.Caching.ICacheProvider, OnlineStore.Infrastructure" mapTo="OnlineStore.Infrastructure.Caching.EntLibCacheProvider, OnlineStore.Infrastructure" />
<!--........-->
</container>
</unity>