我们在开发项目的时候，当我们的业务越来越多的时候，网站所需要的服务器压力也会越来越大，下面就让我们一起来了解一下详解Discuz!NT数据库读写分离方案吧！
目前在Discuz!NT这个产品中，数据库作为数据持久化工具，必定在并发访问频繁且负载压力较大的情况下成 为系统性能的‘瓶颈'。即使使用本地缓存等方式来解决频繁访问数据库的问题，但仍旧会有大量的并发请求要访问动态数据，虽然 SQL2005及2008以上版本中性能不断提升，查询计划和存储过程运行得越来越高效，但最终还是 要面临‘瓶颈'这一问 题。当然这也是许多大型网站不断研究探索各式各样的方案来有效降低数据访问负荷的原 因， 其中的‘读写分离'方案就是一种被广泛采用的方案。
Discuz!NT这个产品在其企业版中提供了对‘读写分离'机制的支持，使对CPU及内存消耗严重的操作(CUD)被 分离到一台或几台性能很高的机器上，而将频繁读取的操作(select)放到几台配置较低的机器上，然后通过‘事务 发布订阅机制'，实现了在多个sqlserver数据库之间快速高效同步数据，从而达到了将‘读写请求'按实际负载 情况进行均衡分布的效果。
下面就简要介绍一下其实现思路。注：有关数据同步的工具已在sqlserver中自带了，可以参考这篇文章。
将相应的数据由Master(主)数据库中‘发布'出来，然后使用推送的方式（注：事务发布可以指定是‘通过主 数据库推送' 还是‘订阅服务器去获取'）发送到订阅它的数据库中，就实现了数据同步功能。
下面就介绍一下如何通过改变既有代码来实现在‘几个从数据库（类似快照）'间进行读取数据的负载均衡。
原有的代码中因为使用了分层机制，所以我们只要在‘数据访问层'动一下心思就可以了。在这里我的一个设 计思路就是不改变已有的数据库访问接口（包括参数等）的前提下，实现底层自动将现有的数据访问操作进行负载 均衡。这样做的好处不用多说了，同时也让这个负载均衡功能与数据访问层相分离，不要耦合的太紧密，同时如果不晓得底层 的实现原理也可以只通过一个开关（后面会介绍），就可以让自己的sql语句自动实现动态负载均衡。
说到这里，我来对照代码进一步阐述：
首先就是(Discuz.Data/DbHelper.cs)代码，主要变动如下(新增方法部分)：
复制代码 代码如下:
/// <summary>
/// 获取使用的数据库(或快照)链接串
/// </summary>
/// <param name="commandText">存储过程名或都SQL命令文本</param>
/// <returns></returns>
public static string GetRealConnectionString(string commandText)
{
if (DbSnapConfigs.GetConfig() != null && DbSnapConfigs.GetConfig().AppDbSnap)
{
commandText = commandText.Trim().ToLower();
if (commandText.StartsWith("select") || ((commandText.StartsWith(BaseConfigs.GetTablePrefix) && UserSnapDatabase(commandText))))
{
DbSnapInfo dbSnapInfo = GetLoadBalanceScheduling.GetConnectDbSnap();
if (DbSnapConfigs.GetConfig().RecordeLog && snapLogList.Capacity > snapLogList.Count)
snapLogList.Add(string.Format("{{'SouceID' : {0}, 'DbconnectString' : '{1}', 'CommandText' : '{2}', 'PostDateTime' : '{3}'}},",
dbSnapInfo.SouceID,
dbSnapInfo.DbconnectString,
commandText.Replace("'",""),
Discuz.Common.Utils.GetDateTime()));
return dbSnapInfo.DbconnectString;
}
}
return ConnectionString;
}
上面的方法将会对传入的sql语句进行分析，找出其中是CUD操作还是SELECT操作，来区别是读还是写操作。而snapLogList列表则是之前所配置的‘事务发布订阅'模式下的相关‘从数据库'(Slave Database)链接串的列表，例如(dbsnap.config文件的DbSnapInfoList节点)：
复制代码 代码如下:
<?xml version="1.0"?>
<DbSnapAppConfig xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<AppDbSnap>true</AppDbSnap>
<WriteWaitTime>1</WriteWaitTime>
<LoadBalanceScheduling>RoundRobinScheduling</LoadBalanceScheduling> --WeightedRoundRobinScheduling
<RecordeLog>false</RecordeLog>
<DbSnapInfoList>
<DbSnapInfo>
<SouceID>1</SouceID>
<Enable>true</Enable>
<DbconnectString>Data Source=DAIZHJ/DNT_DAIZHJ;User ID=sa;Password=123123;Initial Catalog=dnt_snap;Pooling=true</DbconnectString>
<Weight>4</Weight>
</DbSnapInfo>
<DbSnapInfo>
<SouceID>2</SouceID>
<Enable>true</Enable>
<DbconnectString>Data Source=DAIZHJ-PC/2222;User ID=sa;Password=123;Initial Catalog=tabletest;Pooling=true</DbconnectString>
<Weight>3</Weight>
</DbSnapInfo>
</DbSnapInfoList>
</DbSnapAppConfig>
有关相应配置节点和负载均衡算法会在后面提到，这里为了保持文章内容的连续性暂且跳过，下面接着浏览一下上面调用的‘UserSnapDatabase'方法：
复制代码 代码如下:
/// <summary>
/// 是否使用快照数据库
/// </summary>
/// <param name="commandText">查询</param>
/// <returns></returns>
private static bool UserSnapDatabase(string commandText)
{
// 如果上次刷新cookie间隔小于5分钟, 则不刷新数据库最后活动时间
if (commandText.StartsWith(BaseConfigs.GetTablePrefix + "create"))
{
Utils.WriteCookie("JumpAfterWrite", Environment.TickCount.ToString());
return false;
}
else if (!String.IsNullOrEmpty(Utils.GetCookie("JumpAfterWrite")) && (Environment.TickCount - TypeConverter.StrToInt(Utils.GetCookie("JumpAfterWrite"), Environment.TickCount)) < DbSnapConfigs.GetConfig().WriteWaitTime * 1000)
return false;
else if (!commandText.StartsWith(BaseConfigs.GetTablePrefix + "get"))
return false;
return true;
}
该方法的作用很简单，就是当数据库有CUD操作时，通过写cookie的方式向客户端写一个键值‘JumpAfterWrite'，这个键值很重要，就是提供一个标签(flag)来指示：‘当前用户执行cud操作时，页面跳转到其它页面而主数据库还没来得及将数据推送到从数据库'这一情况而造成的‘数据不同步'问题。
举了例子，当在一个版块中‘发表主题'后系统自动跳转到‘显示该主题页面'时，如果主数据库中插入了一个新主题而从数据库没有被及时更新这一主题信息时，就会报‘主题不存在'这个错误。所以这里加了一个设置，就是下面这一行：
复制代码 代码如下:
(Environment.TickCount - TypeConverter.StrToInt(Utils.GetCookie("JumpAfterWrite"), Environment.TickCount)) < DbSnapConfigs.GetConfig().WriteWaitTime * 1000)
它所做的就是确保用户cud操作之后，在规定的时间内还是访问主数据库，当时间超过时，才将当前用户的访问请求(select)均衡到其它从数据库中。
当然，在GetRealConnectionString()方法中，还有一行代码很重要，就是下面这一行：
复制代码 代码如下:
DbSnapInfo dbSnapInfo = GetLoadBalanceScheduling.GetConnectDbSnap();
它的作用就是加载配置文件信息，其中最主要的就是相应的‘负载均衡算法实例'来获取相应的从数据库链接串，下面先看一
下‘静态属性'GetLoadBalanceScheduling的相关信息：
代码
复制代码 代码如下:
/// <summary>
/// 负载均衡调度接口
/// </summary>
private static ILoadBalanceScheduling m_loadBalanceSche;
/// <summary>
/// 初始化负载均衡调度接口实例
/// </summary>
private static ILoadBalanceScheduling GetLoadBalanceScheduling
{
get
{
if (m_loadBalanceSche == null)
{
try
{
m_loadBalanceSche = (ILoadBalanceScheduling)Activator.CreateInstance(Type.GetType(string.Format("Discuz.EntLib.{0}, Discuz.EntLib", DbSnapConfigs.GetConfig().LoadBalanceScheduling), false, true));
}
catch
{
throw new Exception("请检查config/dbsnap.config中配置是否正确");
}
}
return m_loadBalanceSche;
}
}
它主要是通过反射的方法将Discuz.EntLib.dll文件中的相应负载均衡算法实例进行绑定，然后以m_loadBalanceSche这个静态变量进行保存，而m_loadBalanceSche本身就是ILoadBalanceScheduling接口变量，该接口即是相应负载均衡算法的实现接口。同样因为文章内容的连续性，这里先不深挖相应的实现算法，我会在后面进行介绍。下面再来看一下GetRealConnectionString()中还有一段代码，如下：
代码
复制代码 代码如下:
if (DbSnapConfigs.GetConfig().RecordeLog && snapLogList.Capacity > snapLogList.Count)
snapLogList.Add(string.Format("{{'SouceID' : {0}, 'DbconnectString' : '{1}', 'CommandText' : '{2}', 'PostDateTime' : '{3}'}},",
dbSnapInfo.SouceID,
dbSnapInfo.DbconnectString,
commandText.Replace("'",""),
Discuz.Common.Utils.GetDateTime()));
return dbSnapInfo.DbconnectString;
上面代码将当前的负载均衡得到的链接串保存到一个snapLogList列表中，该列表声明如下：
复制代码 代码如下:
List<string> snapLogList = new List<string>(400)
为什么要提供这个列表并进行记录？主要是为了考查负载均衡算法的工作情况，因为在数据访问层获取相应链接串信息并进行记录很不方便，所以我用这个变量记录大约400条‘负载均衡'数据链接串，以便在相应的Discuz.EntLib.ToolKit工具包中进行观察，监视其‘工作情况'。这里我们只要知道通过GetRealConnectionString()方法就实现了对sql语句或存储过程进行分析并进行负载均衡的效果了（注：该操作可能会耗时，所以在DbSnapConfigs中提供了一个开关‘RecordeLog'来进行控制，后面会介绍）。
下面再来简单介绍一下，如何改造DbHelper.cs中原有方法，使其支持负载均衡功能。这里强调一点，就是：
GetRealConnectionString()方法只是造了一个房子，里面的家具还是要自己搬。
而家具就是那些老的方法，比如：
代码
复制代码 代码如下:
public static object ExecuteScalar(DbConnection connection, CommandType commandType, string commandText, params DbParameter[] commandParameters)
{
if (connection == null) throw new ArgumentNullException("connection");
//connection.Close();
connection.ConnectionString = GetRealConnectionString(commandText);//负载均衡改造完成的方法
connection.Open();
// 创建DbCommand命令,并进行预处理
DbCommand cmd = Factory.CreateCommand();
bool mustCloseConnection = false;
PrepareCommand(cmd, connection, (DbTransaction)null, commandType, commandText, commandParameters, out mustCloseConnection);
// 执行DbCommand命令,并返回结果.
object retval = cmd.ExecuteScalar();
// 清除参数,以便再次使用.
cmd.Parameters.Clear();
if (mustCloseConnection)
connection.Close();
return retval;
}
上面的 ‘connection.ConnectionString ='之前绑定的ConnectionString这个静态属性，而这个属性链接的就是‘主数据库'，
这里我们只要将GetRealConnectionString(commandText)赋值给它就可以了，还是那句话，在GetRealConnectionString()就实现了
数据库链接串的负载均衡，呵呵。类似上面的变动在DbHelper.cs还有几处，好在变化不太大，当然更不需要改变原有的数据访问层
（比如IDataProvider.cs文件）了。
其实本文中介绍的数据库层负载均衡实现方法在MYSQL中早有相应的插件实现了，参见这篇文章。
该文章中的LUA脚本实现方式与本文类似，如下：
--发送所有的非事务性SELECT到一个从数据库
复制代码 代码如下:
if is_in_transaction == 0 and packet:byte() == proxy.COM_QUERY and packet:sub(2, 7) == "SELECT" then
local max_conns = -1
local max_conns_ndx = 0
for i = 1, #proxy.servers do
local s = proxy.servers[i]
-- 选择一个拥有空闲连接的从数据库
if s.type == proxy.BACKEND_TYPE_RO and s.idling_connections > 0 then
if max_conns == -1 or s.connected_clients < max_conns then
max_conns = s.connected_clients
max_conns_ndx = i
end
end
end
.....
接着，我再介绍一下相应的配置文件和负载均衡算法的实现情况：）
配置文件（比如：Discuz.EntLib.ToolKit/config/dbsnap.config）：
代码
复制代码 代码如下:
<?xml version="1.0"?>
<DbSnapAppConfig xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<AppDbSnap>true</AppDbSnap>
<WriteWaitTime>1</WriteWaitTime>
<LoadBalanceScheduling>RoundRobinScheduling</LoadBalanceScheduling> --WeightedRoundRobinScheduling
<RecordeLog>false</RecordeLog>
<DbSnapInfoList>
<DbSnapInfo>
<SouceID>1</SouceID>
<Enable>true</Enable>
<DbconnectString>Data Source=DAIZHJ/DNT_DAIZHJ;User ID=sa;Password=123123;Initial Catalog=dnt_snap;Pooling=true</DbconnectString>
<Weight>4</Weight>
</DbSnapInfo>
<DbSnapInfo>
<SouceID>2</SouceID>
<Enable>true</Enable>
<DbconnectString>Data Source=DAIZHJ-PC/2222;User ID=sa;Password=123;Initial Catalog=tabletest;Pooling=true</DbconnectString>
<Weight>3</Weight>
</DbSnapInfo>
<DbSnapInfo>
<SouceID>3</SouceID>
<Enable>true</Enable>
<DbconnectString>Data Source=DAIZHJ-PC/333333;User ID=sa;Password=123;Initial Catalog=tabletest;Pooling=true</DbconnectString>
<Weight>2</Weight>
</DbSnapInfo>
<DbSnapInfo>
<SouceID>4</SouceID>
<Enable>true</Enable>
<DbconnectString>Data Source=DAIZHJ-PC/44444444;User ID=sa;Password=123;Initial Catalog=tabletest;Pooling=true</DbconnectString>
<Weight>2</Weight>
</DbSnapInfo>
</DbSnapInfoList>
</DbSnapAppConfig>
上面的DbSnapInfoList就是相应的slave数据库链接列表，其中它的相应节点信息说明如下（Discuz.Config/DbSnapInfo.cs）：
代码
复制代码 代码如下:
[Serializable]
public class DbSnapInfo
{
/// <summary>
/// 源ID,用于唯一标识快照在数据库负载均衡中的信息
/// </summary>
private int _souceID;
/// <summary>
/// 源ID,用于唯一标识快照在数据库负载均衡中的信息
/// </summary>
public int SouceID
{
get { return _souceID; }
set { _souceID = value; }
}
/// <summary>
/// 快照是否有效
/// </summary>
private bool _enable;
/// <summary>
/// 是否有效
/// </summary>
public bool Enable
{
get { return _enable; }
set { _enable = value; }
}
/// <summary>
/// 快照链接
/// </summary>
private string _dbConnectString;
/// <summary>
/// 快照链接
/// </summary>
public string DbconnectString
{
get { return _dbConnectString; }
set { _dbConnectString = value; }
}
/// <summary>
/// 权重信息，该值越高则意味着被轮循到的次数越多
/// </summary>
private int _weight;
/// <summary>
/// 权重信息，该值越高则意味着被轮循到的次数越多
/// </summary>
public int Weight
{
get { return _weight; }
set { _weight = value; }
}
}
当然DbSnapAppConfig作为DbSnapInfo列表的容器，其结构如下：
代码
复制代码 代码如下:
[Serializable]
public class DbSnapAppConfig : Discuz.Config.IConfigInfo
{
private bool _appDbSnap;
/// <summary>
/// 是否启用快照，如不使用，则即使DbSnapInfoList已设置有效快照信息也不会使用。
/// </summary>
public bool AppDbSnap
{
get { return _appDbSnap; }
set { _appDbSnap = value; }
}
private int _writeWaitTime = 6;
/// <summary>
/// 写操作等待时间(单位:秒), 说明:在执行完写操作之后，在该时间内的sql请求依旧会被发往master数据库
/// </summary>
public int WriteWaitTime
{
get { return _writeWaitTime; }
set { _writeWaitTime = value; }
}
private string _loadBalanceScheduling = "WeightedRoundRobinScheduling";
/// <summary>
/// 负载均衡调度算法，默认为权重轮询调度算法 http://www.pcjx.com/Cisco/zhong/209068.html
/// </summary>
public string LoadBalanceScheduling
{
get { return _loadBalanceScheduling; }
set { _loadBalanceScheduling = value; }
}
private bool _recordeLog = false;
/// <summary>
/// 是否记录日志
/// </summary>
public bool RecordeLog
{
get { return _recordeLog; }
set { _recordeLog = value; }
}
private List<DbSnapInfo> _dbSnapInfoList;
/// <summary>
/// 快照轮循列表
/// </summary>
public List<DbSnapInfo> DbSnapInfoList
{
get { return _dbSnapInfoList; }
set { _dbSnapInfoList = value; }
}
}
通过这两个配置文件，就可以实现对数据访问层负载均衡的灵活配置了，不过上面的DbSnapAppConfig还有一个非常重要的
属性没有介绍清楚，就是‘LoadBalanceScheduling'，其接口声明如下：
代码
复制代码 代码如下:
/// <summary>
/// 负载均衡调度接口
/// </summary>
public interface ILoadBalanceScheduling
{
/// <summary>
/// 获取应用当前负载均衡调度算法下的快照链接信息
/// </summary>
/// <returns></returns>
DbSnapInfo GetConnectDbSnap();
}
它就是负载均衡算法的实现接口，为了便于说明在Discuz.EntLib中内置的两个负载均衡算法的实现情况，请先看下图：
内置的两个负载均衡算法，一个是RoundRobinScheduling，即轮叫调度（Round Robin Scheduling）算法，它的实现比较简单，就是对从数据库链接列表的依次遍历，如下：
代码
复制代码 代码如下:
/// <summary>
/// 轮叫调度（Round Robin Scheduling）算法
/// </summary>
public class RoundRobinScheduling : ILoadBalanceScheduling
{
private static object lockHelper = new object();
/// <summary>
/// 当前的快照索引和权重信息
/// </summary>
static int curentSnapIndex = 0;
static RoundRobinScheduling()
{}
public DbSnapInfo GetConnectDbSnap()
{
lock (lockHelper)
{
if (curentSnapIndex >= DbSnapConfigs.GetEnableSnapList().Count)
curentSnapIndex = (curentSnapIndex) % DbSnapConfigs.GetEnableSnapList().Count;
return DbSnapConfigs.GetEnableSnapList()[curentSnapIndex++];
}
}
}
而另一种负载均衡算法就相对负载了，不过它也更符合实际的应用场景，它使用了权重的方法来让性能优良的机器分到
更多的任务来均衡整个方案的性能，即权重轮询调度算法，实现代码如下：
代码
复制代码 代码如下:
/// <summary>
/// 权重轮询调度算法
/// http://www.pcjx.com/Cisco/zhong/209068.html
/// http://id-phatman.spaces.live.com/blog/cns!CA763CA8DB2378D1!627.entry
/// </summary>
public class WeightedRoundRobinScheduling : ILoadBalanceScheduling
{
private static object lockHelper = new object();
/// <summary>
/// 快照的权重列表
/// </summary>
static List<int> snapWeightList = new List<int>();
/// <summary>
/// 当前的快照索引和权重信息
/// </summary>
static int curentSnapIndex, currentWeight;
/// <summary>
/// 快照权重列表中最大的权重值和最大公约数
/// </summary>
static int maxWeight, gcd;
static WeightedRoundRobinScheduling()
{
curentSnapIndex = -1;
currentWeight = 0;
snapWeightList = GetSnapWeightList();
maxWeight = GetMaxWeight(snapWeightList);
gcd = GCD(snapWeightList);
}
/// <summary>
/// 获取应用当前负载均衡调度算法下的快照链接信息
/// </summary>
/// <returns></returns>
public DbSnapInfo GetConnectDbSnap()
{
lock (lockHelper)
{
DbSnapInfo current = RoundRobinScheduling();
if (current != null)
return current;
else
return DbSnapConfigs.GetEnableSnapList()[0];
}
}
/// <summary>
/// 获取快照权重的列表
/// </summary>
/// <returns></returns>
static List<int> GetSnapWeightList()
{
List<int> snapWeightList = new List<int>();
foreach (DbSnapInfo dbSnapInfo in DbSnapConfigs.GetEnableSnapList())
{
snapWeightList.Add(dbSnapInfo.Weight);
}
return snapWeightList;
}
/// <summary>
/// 权重轮询调度算法
/// </summary>
static DbSnapInfo RoundRobinScheduling()
{
while (true)
{
curentSnapIndex = (curentSnapIndex + 1) % DbSnapConfigs.GetEnableSnapList().Count;
if (curentSnapIndex == 0)
{
currentWeight = currentWeight - gcd;
if (currentWeight <= 0)
{
currentWeight = maxWeight;
if (currentWeight == 0)
return null;
}
}
if (DbSnapConfigs.GetEnableSnapList()[curentSnapIndex].Weight >= currentWeight)
return DbSnapConfigs.GetEnableSnapList()[curentSnapIndex];
}
}
/// <summary>
/// 获取最大权重
/// </summary>
/// <param name="snapList"></param>
/// <returns></returns>
static int GetMaxWeight(List<int> snapWeightList)
{
int maxWeight = 0;
foreach (int snapWeight in snapWeightList)
{
if (maxWeight < snapWeight)
maxWeight = snapWeight;
}
return maxWeight;
}
/// <summary>
/// 获取权重的最大公约数
/// </summary>
/// <returns></returns>
static int GCD(List<int> snapWeightList)
{
// 排序，得到数字中最小的一个
snapWeightList.Sort(new WeightCompare());
int minNum = snapWeightList[0];
// 最大公约数肯定大于等于1，且小于等于最小的那个数。
// 依次整除，如果余数全部为0说明是一个约数，直到打出最大的那个约数
int gcd = 1;
for (int i = 1; i <= minNum; i++)
{
bool isFound = true;
foreach (int snapWeight in snapWeightList)
{
if (snapWeight % i != 0)
{
isFound = false;
break;
}
}
if (isFound)
gcd = i;
}
return gcd;
}
/// <summary>
/// 实现IComparer接口,用于对数字列表进行排序
/// </summary>
private class WeightCompare : System.Collections.Generic.IComparer<int>
{
public int Compare(int weightA, int weightB)
{
return weightA - weightB;
}
}
}
到这里，主要的功能代码就介绍的差不多了，我们可以通过对dbsnap.config的相应节点配置，来灵活定制我们的负载均衡方案。同时，对一般开发者而言，这种架构是透明的，大家可以完全在不了解它的情况下开发自己的数据访问功能，并通过相应开关来让自己的代码支持均衡负载。
当然这个方案还有一些没考虑到的问题比如：
1.对‘主从数据库的健康度检查'，即如果主或从数据库出现故障的时候该如何处理，当然在sqlserver中还提供了镜像功能来解决类似问题，所以它也可做为一个备选方案。
2.当主数据库被发布出去后，主数据库的表和存储过程就会被‘锁定'，其不允许被再次修改了，所以还要继续研究如何解决这一问题。
上述是错新技术频道小编为大家带来的详解Discuz!NT数据库读写分离方案，我们可以在实际操作过程中按上述介绍进行操作，如有任何疑问，欢迎在下方进行留言，我们将第一时间为大家解决。
