一、需求

移動端系統裡有使用者和文章，文章可設定許可權對部分使用者開放。現要實現的功能是，使用者瀏覽自己能看的最新文章，並可以上滑分頁檢視。

二、資料庫表設計

 涉及到的資料庫表有：使用者表TbUser、文章表TbArticle、使用者可見文章表TbUserArticle。其中，TbUserArticle的結構和資料如下圖，欄位有：自增長主鍵id、使用者編號uid、文章編號aid。

自增長主鍵和分散式增長主鍵如何選：

TbUserArticle的主鍵是自增id，它有個缺陷是，當你的資料庫有主從複製時，主從庫的自增可能因死鎖等原因導致不同步。不過，我們可以知道，這裡的TbUserArticle的主鍵id不會做其它表的外來鍵，所以可以是自增id。不像使用者表的主鍵，它就不能用自增id，因為使用者表主鍵(uid)會經常出現在其它表中，當主從庫自增不一致時，很多有uid欄位的表資料在從庫中就不正確了。使用者表主鍵最好是用分散式增長主鍵演算法生成的id（比如Snowflake雪花演算法）。

那麼你可能就要說了，TbUserArticle的主鍵為什麼不直接用雪花演算法產生，不管有沒有用，先讓主從庫主鍵值一致總是有恃無恐。要知道，雪花演算法產生的id一般是18位，而redis的zset的score是double型別，只能表達到16位"整數"部分(精確的說是9007199254740992=2的53次方)。因此，TbUserArticle的主鍵選擇自增id。

主鍵一般都要選自增id或分散式增長id，這種主鍵好處多多，它符合自增長(物理儲存時都是在末尾追加資料，減少資料移動)、唯一性、長度小、查詢快的特性，是聚集索引的很好選擇。

三、redis快取設計-zset

zset的作法及其優點說明：

1.zset的score倒序取數可以很好的滿足取最新資料的需求。

2.用TbUserArticle的文章編號當value，用自增長id當score。自增id的唯一性可很方便的取下一頁資料，直接取小於上次最後一筆的score即可（用lastScore表示）。而如果用文章的時間做score，則要考慮兩筆文章的時間是同分同秒問題，當lastScore落在同分同秒的兩篇文章之間時，就尷尬了，雖然有解，但麻煩了一點。有時的場景你用不了自增id當score，只能用文章時間，那怎麼解決呢，方案就是當是同分同秒時，再根據文章編號做比較就好了，zset的score相同時，也是再根據value排序的。

3.當新增或重新新增一項時，zset也會保持score排序。而如果用的是redis的list，一般就得從db過載快取，新增進來的資料項就算是最新的，也不敢直接新增到list第一筆，因為併發情況下，保證不了最新就是在第一筆；至於重新新增進非最新項，那更是要從db取數重新裝載快取（一般是直接刪除快取，要用的時候才裝載）。

4.第一次從db載入資料到zset時，可只取前N筆到zset。因為我們移動端的資料瀏覽，一般是隻看最新N筆，當看到昨天瀏覽過的資料一般就不會再往下瀏覽。

5.控制zset為固定長度，防止一直增長，一是減少快取開銷，二是佇列長度越短操作效能越高。而且redis服務端有兩個引數：zset-max-ziplist-entries(zset佇列長度，預設值128)和 zset-max-ziplist-value(zset每項大小，預設值64位元組)，它們的作用是，當zset長度小於128，且每個元素的大小小於64位元組時，會啟用ziplist(壓縮雙向連結串列)，它的記憶體空間可以減少8倍左右，而且操作效能也更快。如果不滿足這兩個條件則是普通的skiplist（跳躍表）。另，資料結構hash和list預設長度是512。如果系統有100萬個使用者，每個使用者都有自己的佇列快取，那麼使用ziplist將節省非常大的記憶體空間，並提升很大的效能。

注意，當從zset移除一項資料，則看場景是否需要清空佇列。否則有可能新增進來了一項很舊的資料，它會跑到快取佇列最底部，如果此舊資料比db中未進佇列的資料還舊，那麼佇列中的資料就不正確了。此時，使用者滑到快取最後一頁時，就有可能瀏覽到這項不正確的資料，為什麼是“有可能”，因為當取到zset最後一筆，很可能不夠一頁，而不夠一頁就會從db直接取一頁；而當又新增進一項新資料，這項舊資料就會被T出佇列（因為佇列保持固定長度）。最佳方案是搞個臨界值處理此問題。 而如果新增到zset的資料都是最新資料，則不會有此問題。一般是這種情況，才可以用自增id當score。

 當用唯一主鍵id做score時，這可是非常有用，你可以直接根據id定位到項了，至於如何大用它，我會再出篇部落格。

四、程式碼實現 

從redis快取按頁取數一般要考慮的點：

1.當根據cacheKey未取到資料時（可能是快取過期了導致redis無此cacheKey資料），則觸發過載資料(reload)：從db取limit N筆資料，裝載到redis zset佇列中，並直接取N筆的第一頁資料返回；
2.如果db本身也無對應資料，則新增"no_db_mark"標識到cacheKey佇列中，下次請求則不會再觸發db過載資料；
3.當取到快取末尾時，從db取一頁資料直接返回。這種情況是很少的，要根據業務場景合理規劃快取長度。

上程式碼：

程式碼註釋比較詳細和有用，請直接看程式碼。其中，批量新增資料到zset的函式AddItemsToZset很有用，它使用lua一次性新增資料到zset（注意，使用lua時，要保證lua執行快，否則它會阻塞其它命令的執行），經測試：AddItemsToZset新增1w筆資料，只需要39ms；10w筆需要448ms。因為我們只取前N筆資料到快取，因此一般不會新增超過1w筆。

  1     /// <summary>
  2     /// 分頁取數幫助類
  3     /// </summary>
  4     public class PageDataHelper
  5     {
  6         public readonly static string NoDbDataMark = "no_db_data";//在zset中標識db也無資料
  7         public static RedisHandle RedisClient = new RedisHandle();//redis操作物件示例
  8         public static DbHandleBase DbHandle = new SqlServerHandle("Data Source=.;Initial Catalog=Test;User Id=sa;Password=123ewq;");//db操作物件示例
  9         /// <summary>
 10         /// 按頁取數。返回文章編號列表。
 11         /// </summary>
 12         /// <param name="lastInfo">上一頁最後一筆的score，如果為空，則說明是取第一頁。</param>
 13         /// <param name="getPast">true，使用者上滑瀏覽下一頁資料；false，使用者上滑瀏覽最新一頁資料</param>
 14         /// <returns>返回key-value列表，key就是文章編號，value就是自增id(可用於lastScore)</returns>
 15         public static IDictionary<string, double> GetUserPageData(string uid, int pageSize, string lastInfo, bool getPast)
 16         {
 17             long lastScore = 0;
 18             //1.解析lastInfo資訊。->getPast為false，則固定取最新第一頁資料，不用解析。lastInfo為空，則也不用解析，預設第一頁
 19             if (getPast && !string.IsNullOrWhiteSpace(lastInfo))
 20             {
 21                 lastScore = long.Parse(lastInfo);//外層有try..catch..
 22             }
 23             string cacheKey = $"usr:art:{uid}";
 24             bool isFirstPage = lastScore <= 0;
 25             using (IRedisClient redis = RedisClient.GetRedisClient())
 26             {
 27                 if (isFirstPage)
 28                 {
 29                     //2.第一頁取數
 30                     var items = redis.GetRangeWithScoresFromSortedSetDesc(cacheKey, 0, pageSize - 1);
 31                     if (items.Count == 0)
 32                     {
 33                         //2.1 無資料時，則從db reload資料
 34                         items = ReloadDataToRedis(redis, cacheKey, uid, pageSize);
 35                         if (items.Count == 0 && pageSize > 0)
 36                         {
 37                             //如果db中也無資料，則向zset中新增一筆NoDbDataMark標識
 38                             redis.AddItemToSortedSet(cacheKey, NoDbDataMark, double.MaxValue);
 39                         }
 40                     }
 41                     else if (items.Count == 1 && items.ContainsKey(NoDbDataMark))
 42                     {
 43                         //2.2如果取到的是NoDbDataMark標識，則說明是空資料，則要Clear，返回空列表
 44                         items.Clear();
 45                     }
 46                     //設定快取有效期，要根據業務場景合理設定快取有效期，這邊以7天為例。
 47                     redis.ExpireEntryIn(cacheKey, new TimeSpan(7, 0, 0, 0));
 48                     //2.3 第一頁，有多少就返回多少資料。資料如果不夠一頁，說明本身資料不夠。
 49                     return items;
 50                 }
 51                 else
 52                 {
 53                     //3.第二頁(及之後)取數
 54                     var items = GetPageDataByLastScoreFromRedis(redis, cacheKey, pageSize, lastScore);
 55                     if (items.Count < pageSize)
 56                     {
 57                         //3.1 如果取不夠資料時，就到db取。如果db也不能取到一頁資料，前端會顯示無更多資料，不會一直db取。
 58                         return GetPageDataByLastScoreFromDb(uid, pageSize, lastScore);
 59                     }
 60                     //3.2 如果快取資料足夠，則返回快取的資料。
 61                     return items;
 62                 }
 63             }
 64         }
 65         public static Dictionary<string, double> ReloadDataToRedis(IRedisClient redis, string cacheKey, string uid, int pageSize, string bizId = "")
 66         {
 67             //1.db取數 取top 1000筆資料。不需要全取到快取。
 68             IEnumerable<dynamic> models;
 69             using (var conn = DbHandle.CreateConnectionAndOpen())
 70             {
 71                 var sql = $"select top 1000 id,aid from TbUserArticle where uid=@uid order by id desc;";// limit 1000;";
 72                 models = conn.Query<dynamic>(sql, new { uid = uid });
 73             }
 74             if (models.Count() <= 0) return new Dictionary<string, double>();
 75             //2.資料載入到redis快取。
 76             var itemsParam = new Dictionary<string, double>();
 77             foreach (dynamic model in models)
 78             {
 79                 itemsParam.Add((string)model.aid, (double)model.id);
 80             }
 81             //使用lua一次性新增資料到快取。lua語句要執行快，經測試新增1w筆資料，只需要39ms；10w筆需要448ms。因為sql中有limit，因此一般不會新增超過1w筆。
 82             //因為是原子性操作、並且是zset結構，這邊不需要加鎖。db取到資料應第一時間載入到redis。
 83             AddItemsToZset(redis, cacheKey, itemsParam, true, true);
 84             if (pageSize <= 0) return null;
 85             //3.直接由models返回第一頁資料。
 86             return models.Take(pageSize).ToDictionary(x => (string)x.aid, y => (double)y.id);
 87         }
 88 
 89         public static Dictionary<string, double> GetPageDataByLastScoreFromDb(string uid, int pageSize, double lastScore)
 90         {
 91             //db取一頁資料。
 92             var sql = $"select top {pageSize} id,aid from TbUserArticle where uid=@uid and id<{lastScore}order by id desc;";// limit {pageSize};";
 93             using (var conn = DbHandle.CreateConnectionAndOpen())
 94             {
 95                 return conn.Query<dynamic>(sql, new { uid = uid }).ToDictionary(x => (string)x.aid, y => (double)y.id);
 96             }
 97         }
 98         #region 通用函式
 99         /// <summary>
100         /// ZSet第一頁之後的取數，從lastScore開始取pageSize筆資料(第一頁之後才有lastScore)。
101         /// 使用lua，保證原子性操作。
102         /// </summary>
103         public static Dictionary<string, double> GetPageDataByLastScoreFromRedis(IRedisClient redis, string zsetKey, int pageSize, double lastScore)
104         {
105             //ZREVRANGEBYSCORE: from lastScore to '-inf'.
106             var luaBody = @"local sets = redis.call('ZREVRANGEBYSCORE', KEYS[1], ARGV[1], '-inf', 'WITHSCORES');
107             local result = {};
108             local index=0;
109             local pageSize=ARGV[2]*1;
110             local lastScore=ARGV[1]*1;
111             for i = 1, #sets, 2 do 
112                 if index>=pageSize then
113                     break;
114                 end
115                 if (lastScore>sets[i+1]*1) then
116                     table.insert(result, sets[i]);
117                     table.insert(result, sets[i+1]);
118                     index=index+1;
119                 end
120             end
121             return result";
122             //ARGV[1]:lastScore ARGV[2]:pageSize
123             var list = redis.ExecLuaAsList(luaBody, new string[] { zsetKey }, new string[] { lastScore.ToString(), pageSize.ToString() });
124             var result = new Dictionary<string, double>();
125             for (var i = 0; i < list.Count; i += 2)
126             {
127                 result.Add(list[i], Convert.ToDouble(list[i + 1]));
128             }
129             return result;
130         }
131         /// <summary>
132         /// 新增一項到zset快取中。
133         /// </summary>
134         /// <param name="item">要新增到zset的資料項</param>
135         /// <param name="maxCount">控制zset最大長度，如果為0，則不控制。</param>
136         /// <returns></returns>
137         public static string AddItemToZset(IRedisClient redis, string zsetKey, KeyValuePair<string, double> item, int maxCount = 0)
138         {
139             var items = new Dictionary<string, double>() { { item.Key, item.Value } };
140             return AddItemsToZset(redis, zsetKey, items);
141         }
142         /// <summary>
143         /// 新增多項到zset快取中。
144         /// </summary>
145         /// <param name="items">要新增到zset的資料列表</param>
146         /// <param name="hasCacheExpire">快取zsetKey是否有設定快取有效期。如果有設定快取有效期，則當快取中無資料時，可能是快取過期；而如果快取無有效期，快取中無資料，就是db和快取都無資料</param>
147         /// <param name="isReload">是否是reload情況，true過載情況；false追加</param>
148         /// <param name="maxCount">控制zset最大長度，如果為0，則不控制。
149         /// 一般不用控制，只在db取數reload時增加limit，以避免全量進快取；
150         /// 如果當前zsetKey是常用快取，會一直暴漲，則才要控制zset長度。</param>
151         /// <returns></returns>
152         public static string AddItemsToZset(IRedisClient redis, string zsetKey, Dictionary<string, double> items, bool hasCacheExpire = true
153             , bool isReload = false, int maxCount = 0)
154         {
155             //!isReload，是因為如果isReload=true情況無資料，則也要進來過載佇列為無資料(即，如果之前有資料要過載為無資料)
156             if (!isReload && items.Count <= 0) return null;
157             var argArr = new List<string>(items.Count * 2 + 2);//lua引數陣列
158             //var hasCacheExpire = cacheValidTime != null;
159             //第一個lua引數是hasCacheExpire
160             argArr.Add(hasCacheExpire ? "1" : "0");
161             //第二個lua引數是maxCount
162             argArr.Add(maxCount.ToString());
163             //組合lua其它引數列表:ZADD的引數
164             foreach (var item in items)
165             {
166                 //Add score。 //ZADD KEY_NAME SCORE1 VALUE1
167                 argArr.Add(item.Value.ToString());
168                 argArr.Add(item.Key);
169             }
170             #region lua
171             /*
172             * 以下lua命令說明。
173             * 1.ZREVRANGE從大到小取第一筆資料firstMark；
174             * 2.快取有設定有效期時(hasCacheExpire=1)，如果第一筆資料firstMark為nil，則說明列表是空(失效key、未生成key)，則不做任何處理，直接返回字串not_exist_key。因為可能是使用者失效資料，使用者長期未訪問，則不新增，後繼來訪問時過載資料。
175             * 3.如果firstMark標識為no_db_data，則是被api標識為db沒資料，而此時因要ZADD資料進來，因此要把此標識刪除。其中，ZREMRANGEBYRANK從小到大刪除，-1是倒數第一筆。
176             * 4.ZADD資料進來
177             * 5.KeepLength保持佇列長度操作。如果佇列長度(由ZCARD獲取)超過指定的maxCount，則從佇列第一筆開始刪除多餘元素，即score最小開始刪除。
178             * 6.maxCount為>0才KeepLength。返回數值:curCount - maxCount。(可以用返回值簡單算出隊列當前長度curCount)。如果返回值小於等於0則說明沒有觸發刪除操作。
179             * 7.maxCount為<=0時，直接返回'no_remove'。
180             */
181             //清空原來，重新載入資料的情況
182             const string reloadLua = "redis.call('DEL', KEYS[1]) ";
183             //追加資料到zset的情況
184             const string addToLua =
185             @"local firstMark = redis.call('ZREVRANGE',KEYS[1],0,0);
186             local hasCacheExpire=ARGV[1]*1;
187             if hasCacheExpire==1 and firstMark and firstMark[1]==nil then
188                 return 'not_exist_key';
189             end
190             if firstMark and firstMark[1]=='{0}' then
191                 redis.call('ZREMRANGEBYRANK', KEYS[1], -1,-1);
192             end";
193             const string constAllLua =
194             @"{0}
195             for i=3, #ARGV, 2 
196                 do redis.call('ZADD', KEYS[1], ARGV[i], ARGV[i+1]);  
197             end
198             local maxCount=ARGV[2]*1;
199             if maxCount>0 then
200               local curCount= redis.call('ZCARD', KEYS[1]);
201               local removeCount=curCount - maxCount;
202               if removeCount>0 then
203                 redis.call('ZREMRANGEBYRANK', KEYS[1], 0,removeCount-1);    
204               end  
205               return removeCount;
206             end
207             return 'no_remove';";
208             #endregion
209             var luaBody = string.Format(constAllLua, isReload ? reloadLua : string.Format(addToLua, NoDbDataMark));
210             var luaResult = redis.ExecLuaAsString(luaBody, new string[] { zsetKey }, argArr.ToArray());
211             return luaResult;
212         }
213         #endregion
214     }

返回key-value列表，key就是文章編號，value就是自增id(可用於lastSc