缓存子系统如何设计(Cachable tag, Memcache/redis support, xml config support, LRU/LFU/本地缓存命中率)

2022-10-22,,,

大家对这段代码肯定很熟悉吧:

public List<UserInfo> SearchUsers(string userName)
{
string cacheKey=string.Format("SearchUsers_{0}", userName);
List<UserInfo> users = cache.Find(cacheKey) as List<UserInfo>;
if (users == null)
{
users = repository.GetUsersByUserName(userName);
cache.Set(cacheKey, users);
}
return users;
} class HttpRuntimeCache
{
public object Find(string key)
{
return HttpRuntime.Cache[key];
}
public void Set(string key, object value)
{
HttpRuntime.Cache[key] = value;
}
}

导致了如下这些问题:

    业务逻辑函数中引入了很多无关的缓存代码,导致DDD模型不够纯
    更换缓存Provider不方便
    加入缓存冗余机制不方便
    没办法同时使用多个缓存系统
    缓存大对象出现异常,比如Memcache有1M的value限制

有诸多问题,因此我们需要引入缓存子系统来解决上述问题,带来的好处:

    DDD模型更加纯
    具体的Cache实现机制可以很灵活,比如HttpRuntimeCache, Memcache, Redis可以同时使用
    加入了Cache冗余机制,不会由于某一台Memcache或者Redis down机导致系统速度很慢,实际上,系统还是会保持飞快(除非backup也down了的情况)
    开发人员更加致力于核心业务,不会分散注意力
    缓存位置透明化,都会在xml配置文件中进行配置

解决方案,要用到这2篇文章的技术:C# 代理应用 - Cachable 和 聊聊Memcached的应用。

主要的思路分2个:

模型端:通过代理来嵌入AOP方法,来判断是否需要缓存,有缓存value则直接返回value;缓存value的写入是通过AOP的后置方法写入的,因此不需要在业务函数中写代码,当然也支持代码调用。

Cache核心对象:这个对象要解决一致性hash算法、cache value大对象分解功能、冗余机制

代理嵌入AOP的方法,已经在这篇文章中说明了 C# 代理应用 - Cachable,有兴趣的看看,这里就不说了,我们来主要看看CacheCoordinator对象的实现

结构图如下:

先来看看UML图:

CacheCore代码(算法核心):

public class CacheCore
{
private ICacheCoordinator cacheProvider = null;
public CacheCore(ICacheCoordinator cacheProvider)
{
this.cacheProvider = cacheProvider;
} public void Set(string location, string key, object value)
{
AssureSerializable(value);
string xml = Serializer2XMLConvert(value);
CacheParsedObject parsedObj = new CacheParsedObject(); string classType = string.Format("{0}", value.GetType().FullName);
if (xml.Length > CacheConfig.CacheConfiguration.MaxCacheEntitySize)
{
/*
key:1@3@ConcreteType
key_1:subvalue1
key_2:subvalue2
key_3:subvalue3
*/
//拆分成更小的单元
int splitCount = xml.Length / CacheConfig.CacheConfiguration.MaxCacheEntitySize;
if (CacheConfig.CacheConfiguration.MaxCacheEntitySize * splitCount < xml.Length)
splitCount++;
parsedObj.MainObject = new KeyValuePair<string, string>(key, string.Format("1@{0}@{1}", splitCount, classType));
for (int i = ; i < splitCount;i++ )
{
if (i == splitCount - ) //最后一段,直接截取到最后,不用给出长度
parsedObj.SplittedElements.Add(xml.Substring(i * CacheConfig.CacheConfiguration.MaxCacheEntitySize));
else //其他,要给出长度
parsedObj.SplittedElements.Add(xml.Substring(i * CacheConfig.CacheConfiguration.MaxCacheEntitySize, CacheConfig.CacheConfiguration.MaxCacheEntitySize));
}
}
else
{
/*
key:1@1@ConcreteType
key_1:value
*/
parsedObj.MainObject = new KeyValuePair<string, string>(key, string.Format("1@1@{0}", classType));
parsedObj.SplittedElements.Add(xml);
} //针对CacheParsedObject进行逐项保存
this.cacheProvider.Put(parsedObj.MainObject.Key, parsedObj.MainObject.Value);
int curIndex = ;
foreach(string xmlValue in parsedObj.SplittedElements)
{
curIndex++;
string tkey=string.Format("{0}_{1}", parsedObj.MainObject.Key, curIndex);
this.cacheProvider.Put(tkey, xmlValue);
}
} public object Get(string location, string key)
{
string mainObjKeySetting = (string)cacheProvider.Get(key);
if (mainObjKeySetting == null || mainObjKeySetting.Length == )
return null; string classType;
CacheParsedObject parsedObj;
GetParsedObject(key, mainObjKeySetting, out classType, out parsedObj); string xmlValue=string.Empty;
parsedObj.SplittedElements.ForEach(t=>xmlValue+=t); using (StringReader rdr = new StringReader(xmlValue))
{
//Assembly.Load("Core");
Type t = Type.GetType(classType);
XmlSerializer serializer = new XmlSerializer(t);
return serializer.Deserialize(rdr);
}
} public void Remove(string location, string key)
{
string mainObjKeySetting = (string)cacheProvider.Get(key);
if (mainObjKeySetting == null || mainObjKeySetting.Length == )
return; string classType;
CacheParsedObject parsedObj;
GetParsedObject(key, mainObjKeySetting, out classType, out parsedObj); int i = ;
parsedObj.SplittedElements.ForEach(t => this.cacheProvider.Remove(string.Format("{0}_{1}", parsedObj.MainObject.Key, i++)));
this.cacheProvider.Remove(parsedObj.MainObject.Key);
}
private void GetParsedObject(string key, string mainObjKeySetting, out string classType, out CacheParsedObject parsedObj)
{
int from = , end = ;
classType = string.Empty;
if (mainObjKeySetting.IndexOf('@') > )
{
end = int.Parse(mainObjKeySetting.Split('@')[]);
classType = mainObjKeySetting.Split('@')[];
} parsedObj = new CacheParsedObject();
parsedObj.MainObject = new KeyValuePair<string, string>(key, string.Format("1@{0}@{1}", end, classType));
for (int i = from; i <= end; i++)
parsedObj.SplittedElements.Add((string)this.cacheProvider.Get(string.Format("{0}_{1}", parsedObj.MainObject.Key, i)));
}
private string Serializer2XMLConvert(object value)
{
using (StringWriter sw = new StringWriter())
{
XmlSerializer xz = new XmlSerializer(value.GetType());
xz.Serialize(sw, value);
return sw.ToString();
}
}
private void AssureSerializable(object value)
{
if (value == null)
throw new Exception("cache object must be Serializable");
if (value.GetType().GetCustomAttributes(typeof(SerializableAttribute), true).Count()<=)
throw new Exception("cache object must be Serializable");
}
}

下面是CacheCoordinator的代码,这个类的加入目的是要加入缓存的冗余机制:

class CacheCoordinator : ICacheCoordinator
{
CacheServerWrapper backupCacheServer = new CacheServerWrapper(CacheConfig.CacheConfiguration.BackupCacheServer);
CacheServersWrapper peerCacheServer = new CacheServersWrapper(CacheConfig.CacheConfiguration.PeerCacheServers); public void Put(string key, object value)
{
peerCacheServer.Put(key, value);
backupCacheServer.Put(key, value); //缓存冗余
} public object Get(string key)
{
object o=peerCacheServer.Get(key);
if (o != null)
return o;
return backupCacheServer.Get(key);
} public void Remove(string key)
{
peerCacheServer.Remove(key);
backupCacheServer.Remove(key);
}
}

剩下的就是具体的CacheProvider和CacheProviderWrapper类了:

public class CacheServerWrapper : ICacheExecutor
{
ICacheExecutor executor = null;
private CacheServerInfo configInfo;
public CacheServerWrapper(CacheServerInfo configInfo)
{
this.configInfo = configInfo;
ICacheExecutor tmpExecutor = null;
switch(this.configInfo.ServerType)
{
case CacheServerType.HttpRuntime:
tmpExecutor = new CacheProvider.HttpRuntimeCacheProvider(configInfo);
break;
case CacheServerType.InMemory:
tmpExecutor = new CacheProvider.InMemoryCacheProvider(configInfo);
break;
case CacheServerType.Memcached:
tmpExecutor = new CacheProvider.MemcachedCacheProvider(configInfo);
break;
case CacheServerType.Redis:
tmpExecutor = new CacheProvider.RedisCacheProvider(configInfo);
break;
default:
tmpExecutor = new CacheProvider.HttpRuntimeCacheProvider(configInfo);
break;
}
executor = tmpExecutor;
} public string FullServerAddress
{
get
{
return this.configInfo.FullServerAddress;
}
} public void Put(string key, object value)
{
executor.Put(key, value);
} public object Get(string key)
{
return executor.Get(key);
} public void Remove(string key)
{
executor.Remove(key);
}
}

只贴出Memcache的操作类

class MemcachedCacheProvider : ICacheExecutor
{
private MemcachedClient mc = new MemcachedClient();
private CacheServerInfo configInfo;
public MemcachedCacheProvider(CacheServerInfo configInfo)
{
this.configInfo = configInfo; //初始化池
SockIOPool pool = SockIOPool.GetInstance();
pool.SetServers(new string[] { string.Format("{0}:{1}", configInfo.ServerAddress, configInfo.ServerPort) });//设置连接池可用的cache服务器列表,server的构成形式是IP:PORT(如:127.0.0.1:11211)
pool.InitConnections = ;//初始连接数
pool.MinConnections = ;//最小连接数
pool.MaxConnections = ;//最大连接数
pool.SocketConnectTimeout = ;//设置连接的套接字超时
pool.SocketTimeout = ;//设置套接字超时读取
pool.MaintenanceSleep = ;//设置维护线程运行的睡眠时间。如果设置为0,那么维护线程将不会启动,30就是每隔30秒醒来一次 //获取或设置池的故障标志。
//如果这个标志被设置为true则socket连接失败,将试图从另一台服务器返回一个套接字如果存在的话。
//如果设置为false,则得到一个套接字如果存在的话。否则返回NULL,如果它无法连接到请求的服务器。
pool.Failover = true; pool.Nagle = false;//如果为false,对所有创建的套接字关闭Nagle的算法
pool.Initialize();
}
public void Put(string key, object value)
{
mc.Set(key, value);
} public object Get(string key)
{
return mc.Get(key);
} public void Remove(string key)
{
mc.Delete(key);
}
}

不能忘了可配置性,xml定义及代码如下:

<?xml version="1.0" encoding="utf-8" ?>
<CacheConfig>
<MaxCacheEntitySize>1048576</MaxCacheEntitySize><!--1*1024*1024-->
<PeerCacheServers>
<CacheServer>
<ServerType>InMemory</ServerType>
<ServerAddress>127.0.0.1</ServerAddress>
<ServerPort>11211</ServerPort>
</CacheServer>
<CacheServer>
<ServerType>InMemory</ServerType>
<ServerAddress>127.0.0.1</ServerAddress>
<ServerPort>11212</ServerPort>
</CacheServer>
</PeerCacheServers>
<BackupCacheServer>
<CacheServer>
<ServerType>InMemory</ServerType>
<ServerAddress>127.0.0.1</ServerAddress>
<ServerPort>11213</ServerPort>
</CacheServer>
</BackupCacheServer>
</CacheConfig>

读取配置信息的代码:

public static class CacheConfiguration
{
static CacheConfiguration()
{
Load();
} private static void Load()
{
PeerCacheServers = new List<CacheServerInfo>();
BackupCacheServer = null; XElement root = XElement.Load(System.IO.Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "CacheConfig.xml")); MaxCacheEntitySize = int.Parse(root.Element("MaxCacheEntitySize").Value);
foreach (var elm in root.Element("PeerCacheServers").Elements("CacheServer"))
{
CacheServerInfo srv = new CacheServerInfo();
srv.ServerAddress = elm.Element("ServerAddress").Value;
srv.ServerPort = int.Parse(elm.Element("ServerPort").Value);
srv.ServerType = (CacheServerType)Enum.Parse(typeof(CacheServerType), elm.Element("ServerType").Value);
PeerCacheServers.Add(srv);
}
foreach (var elm in root.Element("BackupCacheServer").Elements("CacheServer"))
{
CacheServerInfo srv = new CacheServerInfo();
srv.ServerAddress = elm.Element("ServerAddress").Value;
srv.ServerPort = int.Parse(elm.Element("ServerPort").Value);
srv.ServerType = (CacheServerType)Enum.Parse(typeof(CacheServerType), elm.Element("ServerType").Value);
BackupCacheServer = srv;
break;
}
if (PeerCacheServers.Count <= )
throw new Exception("Peer cache servers not found.");
if (BackupCacheServer == null)
throw new Exception("Backup cache server not found.");
AssureDistinctFullServerAddress(PeerCacheServers);
} private static void AssureDistinctFullServerAddress(List<CacheServerInfo> css)
{
Dictionary<string, int> map = new Dictionary<string, int>();
foreach(CacheServerInfo csInfo in css)
{
if (map.ContainsKey(csInfo.FullServerAddress))
throw new Exception(string.Format("Duplicated server address found [{0}].", csInfo.FullServerAddress));
else
map[csInfo.FullServerAddress] = ;
}
}
public static int MaxCacheEntitySize { get; set; }
public static List<CacheServerInfo> PeerCacheServers { get; set; }
public static CacheServerInfo BackupCacheServer { get; set; }
}

代码下载

Append New

其实,我们忽略了一些重要的东西:

    如果Memcached, Redis服务器超过了5台以上,通信量上升很快,怎么办?
    由于取数据牵涉到网络I/O操作,因此速度依然比较慢,怎么办?

让我们来解决吧。

把新的UML图贴上(下图中左边红框中的是新增的):

本地缓存替换策略:LFU/LRU,其他的有很多。

EventBus是分布式的,下面有讲为什么要分布式的。

当Domain层需要获取数据时的逻辑:

    先查看本地缓存中是否存在数据副本,存在则立刻返回(也没有网络I/O了)
    没有则去redis/memcached获取,有则返回;并且把数据放入本地cache中
    最后,实在没有数据,就db里取

当Domain层需要更新数据时的逻辑:

    在本地cache中进行更新操作
    更新分布式缓存
    发布分布式事件,通知其他app server的cache manager去主动拉数据到他们本地缓存

看得出来,加入这个新的角色后,能对下面2项有改善作用:

    降低网络间的通信流量
    增大本地缓存的命中率

缓存子系统如何设计(Cachable tag, Memcache/redis support, xml config support, LRU/LFU/本地缓存命中率)的相关教程结束。

《缓存子系统如何设计(Cachable tag, Memcache/redis support, xml config support, LRU/LFU/本地缓存命中率).doc》

下载本文的Word格式文档,以方便收藏与打印。