大家好，我是《江湖X》、《汉家江湖》的服务器主程、运维、客服和背锅临时工，今天给大家分享一下，这两年我们使用Redis的一点浅薄的心得体会，如果能给大家带来一点点帮助，就心满意足了。
（感谢阿里云）
总（fei）结（hua）
Redis是一款强大的工具，优点是响应微秒级，有广播/订阅功能，支持事务（非集群），支持队列、哈希表和哈希集合，缺点是需要手动管理索引，事务不可回滚，原生集群不支持事务等高级功能。
我们在《汉家江湖》服务器组中，“大规模”使用了Redis（几十个阿里云实例，相对我们的体量来说，挺大规模了），主要有如下四类用途：
永久存储
数据缓存
实时消息通知
消息队列
最后一节，附上简单的Redis事务代码。
永久存储
因为Redis支持持久化（rdb和aof），所以在《江湖X》内测的时候，事情还是比较简单的，我们把所有数据全部丢在Redis，每天日活500，怎么弄都行。（《江湖X》是《汉家江湖》的前作，技术路线一脉相承）。并且，我们也没有使用阿里云的Redis，而是直接在实例ECS上自己搭建的（说搭建有点抬举自己，下载双击bat文件就好了）。
这个时候，我们主要使用了Redis的哈希表和哈希集合。《汉家江湖》客户端提交到服务器的玩家数据，是几千张哈希表，我们直接全部原样存进了Redis，然后把一个玩家的所有哈希表的key名，存进一个玩家的哈希集合，作为索引。如下图所示，这是一个玩家的索引集合，IT=Index Table。
大家可以看到，玩家“xh937789#1”的这个索引集合，有1502个值，对应了这个玩家1502个哈希表。而且这还不是一个高级玩家，高级玩家的哈希表，一般在2500到3000左右。这样，《江湖X》7月上线之后，美妙的事情发生了：阿里云ECS的硬盘扛不住了。（说实话，阿里云的硬盘性能是真不行）
等等，Redis不是内存程序么？关硬盘一毛钱关系啊？这是因为，拿redis当永久存储，一定要面临的问题就是持久化，不然你出个恶性bug，连服务器回档都做不到，那就gg思密达了。（科普：rdb是把Redis里面的所有数据，全量拷贝压缩存储放进硬盘；aof类似于mysql的bin日志，保证可以通过aof文件回溯任一状态）
一开始我们天真的开启了rdb和aof两种持久化方式，rdb一小时复制一次全量到硬盘，aof采用每秒一次强制写，这样可以保证回档精度为1秒（回档精度，咳咳）。上线之后第一天，核心玩家疯狂涌入，我默默地把aof的强制写频率设置为操作系统的30秒，这样回档精度就是30秒了。
由于Redis是存着所有玩家数据的，所以随着用户的增多，rdb开销直线上升；随着日活的增加，aof的开销也日日见涨。上线之后第四天，我默默地关闭了aof功能，这样回档精度就是一小时了（rdb一小时一次）。
所以，当时《江湖X》的备份机制就是，每小时有一份全量拷贝，存在ECS的硬盘里。过了大概半个月，每小时的拷贝文件突破了1G的大小；过了大概一个月，突破2G大小，这样一天就是24G，并且还随着时间的增加不断变大，Redis占用的实时内存也突破6G。
当时觉得，再这么下去，服务器要爆炸，我们游戏随时要嗝屁了。
缓存数据
《江湖X》上线一个月之后，我们必须解决Redis快爆炸这个问题，当时有两个思路：
买买买：8核16G配SSD，不行就上16核32G，最后实在不行就滚服。（我们绝不滚服！）
底层重构：引入mysql或者mongodb做存储，redis做缓存。
贫穷限制了我们的想象力，高配实在是买不起买不起，对我们创业者来说简直是抢劫。
但是，贫穷激发了我们的行动力和创造力，不就是重构么，搞！
经过调研，我们排除了mongodb，决定采用mysql来做持久化。（这里安利一下ServiceStack的Ormlite和Redis，非常好用）玩家数据的流程图如下：
经过一个月有条不紊（都是反话）的开发，我们在2016年9月12号上线了新的服务器，并且引入了阿里云的双可用区备份的Mysql，Redis还是自己搭建在实例上；通过以上方案，回档精度为5分钟，万一Redis出问题，也可以在Mysql里面找到5分钟之前的数据；Mysql通过阿里云来保证双实例热备，整个服务器稳定性相较之前的裸奔状态，提升了无数个数量级。实例上的Redis，是关闭了rdb和aof持久化的，所以对硬盘，没有任何负载。
这之后，《汉家江湖》绝大多数数据，都在Mysql上做持久化，Redis只做缓存（包括好友系统、社交系统等）。
少部分有时间限制的数据，我们还是直接用Redis做存储，例如每天自动刷新的一些数据（日常任务，日常活动），每周举行的一些活动（周末玩法等）。
实时消息通知
Redis的消息订阅功能，非常强大，这里安利一下阿里云的256MB的Redis，一个月25块钱，简直是实时消息的利器。《汉家江湖》的公共聊天、帮会聊天、私聊、系统消息、GM命令、诸多玩法实时消息队列，都是基于Redis的消息订阅来做的。（私聊后续还需要改进）
这里没什么多说的，直接上代码比较实在。基于ServiceStack的Redis，实现了一个简单的消息订阅框架。注意，只能实时通知，离线掉线就会收不到。
public class RedisMsg{
//private readonly Dictionary<string, Thread> _subThreads = new Dictionary<string, Thread>();
private readonly Dictionary<string, IRedisPubSubServer> _subServers
= new Dictionary<string, IRedisPubSubServer>();
protected RedisManagerPool RedisPool;
//初始化
public RedisMsg(string redisStr)
{
RedisPool = new RedisManagerPool(redisStr);
}
//初始化
public RedisMsg(RedisManagerPool redisPool)
{
RedisPool = redisPool;
}
//发布消息
public void Pub(string channel, string msg)
{
lock (this)
{
using (var pub = RedisPool.GetClient())
{
pub.PublishMessage(channel, msg);
}
}
}
//订阅
public bool Sub(string channel, Action<string> callback)
{
if (_subServers.ContainsKey(channel))
{
return false;
}
var redisPubSub = new RedisPubSubServer(RedisPool, channel)
{
OnMessage = (recieveChannel, msg) =>
{
callback(msg);
}
}.Start();
_subServers.Add(channel, redisPubSub);
return true;
}
//取消订阅
public void UnSub(string channel)
{
if (!_subServers.ContainsKey(channel)) return;
var server = _subServers[channel];
server.Stop();
_subServers.Remove(channel);
}}消息队列
《汉家江湖》有一个“论剑”玩法，在一个小时内，会有几万场玩家对战，每一场对战，都是在服务器进行计算，然后将计算结果下发。这就需要引入一个消息队列，保证每一条消息必定被且只被消费一次，并且与消费者离线/在线状态无关。
一开始我们使用了阿里云的消息队列MQ，这是一个支持”亿/秒级别“的消息队列，优点很明显，成熟稳定，巨量支撑；缺点就是，坑很多，因为太复杂了，你永远不知道在哪里会出问题，而且百万千万级别的消息，丢了一点，延迟一点问题不大，但是在”论剑“这种玩家实时对战的时候，出一次问题，就是游戏体验的毁灭性打击。
经过连续两个月晚上9点到10点的蹲点，在阿里云售后的帮助下，我们解决了一大堆MQ的问题。
在那之后，我们终于可以长舒一口气了，因为，我们决定放弃阿里的MQ。
不是它不好，而是我们的需求不满足它的设计目的。我们需要的就是实时性、必定到达和幂等，量级在“千/秒”。所以我们用了Redis的List用来做简单的消息队列，完美满足我们的需求，至今半年没有出过任何问题。
发布任务代码：
public int InitTask(string uid, string teamInfoA, string teamInfoB){
if (!_IsInitialedDispatcher) return 0;
using (var db = _mysql.Open())
{
var task = new ComputeTask
{
UID = uid,
Status = (int)ComputeTaskStatus.Initial,
CreateAt = DateTime.Now,
TeamInfoA = teamInfoA,
TeamInfoB = teamInfoB,
UpdateAt = DateTime.Now
};
var id = (int)db.Insert(task, true);
using (var redis = _redis.GetClient())
{
redis.EnqueueItemOnList(QueueListKey, id.ToString());
}
return id;
}}消费任务代码：
public class RedisTaskQueue{
private static readonly Logger _logger = LogManager.GetCurrentClassLogger();
public int ThreadNum = 0;
public int MaxThreadNum = 12;
//0为停止，1为开始
private int _status = 0;
private RedisManagerPool _redis;
private Thread _loopThread;
private string _topic;
private Action<int> _taskHandle;
public string QueueListKey => $"Queue:{_topic}";
public void Init(string topic, int threadNums, string redisStr, Action<int> taskHandle)
{
Output($"begin init, topic:{topic}, num:{threadNums}, redis:{redisStr}");
_topic = topic;
MaxThreadNum = threadNums;
_taskHandle = taskHandle;
_redis = new RedisManagerPool(redisStr);
Output($"call the start function");
Start();
}
public void Stop()
{
Interlocked.CompareExchange(ref _status, 0, 1);
}
public void Start()
{
Interlocked.CompareExchange(ref _status, 1, 0);
_loopThread = new Thread(DoLoop);
_loopThread.Start();
}
public void DoLoop()
{
using (var redis = _redis.GetClient())
{
Output($"开始一次循环");
while (Interlocked.CompareExchange(ref _status, -1, -1) == 1)
{
if (Interlocked.CompareExchange(ref ThreadNum, -1, -1) >= MaxThreadNum)
{
Output($"超过最大线程数，休息一秒");
Thread.Sleep(1000);
}
else
{
var idStr = redis.BlockingDequeueItemFromList(QueueListKey,
TimeSpan.FromSeconds(1));
if (!string.IsNullOrEmpty(idStr))
{
Thread.Sleep(200);
var taskId = Convert.ToInt32(idStr);
Output($"开始计算任务{taskId}");
Task.Run(() =>
{
Interlocked.Increment(ref ThreadNum);
_taskHandle(taskId);
Interlocked.Decrement(ref ThreadNum);
});
}
}
}
_logger.Info($"循环结束");
}
}
private static void Output(string msg)
{
_logger.Info(msg);
}}通过Watch实现事务
Redis的响应是微秒级的，一秒可以处理百万级别的指令。但是，天有不测风云，人有旦夕祸福，总有些人运气特别好，能遇到这种百万分之一的概率。
《汉家江湖》中的帮会Boss战，一般有几十名玩家参加。Boss血量共享，每次玩家提交伤害，都需要修改同一个当前血量字段，这个字段放在redis中。然后，某一天，神奇的事情发生了，某帮会有一名玩家，明明打了Boss，最后没有收到奖励。通过排查日志，他提交伤害的时候，与另外一名玩家冲突了，导致Redis事务失败（论日志的重要性）。下面是改进的Redis事务代码，通过“watch”关键的key值，可以重启事务，来提升事务的成功率。
var bFlag = true;while (bFlag){
var bKill = false;
redis.Watch(infoHKey);
redis.Watch(scoreZKey);
var currentHp = Convert.ToInt32(redis.GetValueFromHash(infoHKey, "CurrentHp"));
if (currentHp <= 0) return false;
if (damage >= currentHp)
{
damage = currentHp;
bKill = true;
}
using (var trans = redis.CreateTransaction())
{
var damage1 = damage;
trans.QueueCommand(r => r.IncrementItemInSortedSet(scoreZKey, userKey, damage1));
trans.QueueCommand(r => r.IncrementValueInHash(infoHKey, "CurrentHp", -damage1));
trans.QueueCommand(r => r.AddItemToSortedSet(timeZKey, userKey, now));
if (trans.Commit()) bFlag = false;
}
if (bKill && !bFlag) SendKillMessage(bossId, bhId); //向redis发送消息，管理服来结算}对于Redis，还只是了解了一点皮毛，欢迎大家指出文中的任何问题，谢谢！
源文：https://zhuanlan.zhihu.com/p/34672726