使用redis的zset实现高效分页查询(附完整代码)

您所在的位置:网站首页 springboot16mybatis与实现分页 使用redis的zset实现高效分页查询(附完整代码)

使用redis的zset实现高效分页查询(附完整代码)

2023-03-21 22:00| 来源: 网络整理| 查看: 265

一、需求

移动端系统里有用户和文章,文章可设置权限对部分用户开放。现要实现的功能是,用户浏览自己能看的最新文章,并可以上滑分页查看。

 

二、数据库表设计

 涉及到的数据库表有:用户表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。那么能不能产生一个16位(具体是53bit)的分布式增长id用于支持zset的score呢,当然也是可以的,因为目前的雪花算法是可以根据实际系统环境压缩bit位的,怎么压缩bit位呢,有许多方案,以后有需要我可以把它写出来。

建议:主键一般都要选自增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最后一笔,很可能不够一页(一页10笔计算的话,90%会取不够一页),而不够一页就会从db直接取一页,从db直接取就不会有这项不正确的数据。而当zset又添加进一项新数据,末端那笔旧数据就会被T出队列(因为队列保持固定长度),zset数据又恢复正确了。不管怎样,这种问题几率虽不高,也是有解决方案,可搞个临界点处理此问题,不细说,否则又是长篇大论,最好的方案就是根据实际场景设计,比如从zset队列移除数据的情况多不多)。而如果添加到zset的数据都是最新数据,则不会有此问题。

 当用唯一主键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笔。

另一个通用有用的函数是GetPageDataByLastScoreFromRedis,它支持从指定的score开始取pageSize笔数据,即支持了zset分页。它是第二页(及之后)的取数,而如果取第一页取数,则直接用redis原生函数即可redis.GetRangeWithScoresFromSortedSetDesc(cacheKey, 0, pageSize - 1);。

/// /// 分页取数帮助类 /// public class PageDataHelper { public readonly static string NoDbDataMark = "no_db_data";//在zset中标识db也无数据 public static RedisHandle RedisClient = new RedisHandle();//redis操作对象示例 public static DbHandleBase DbHandle = new SqlServerHandle("Data Source=.;Initial Catalog=Test;User Id=sa;Password=123ewq;");//db操作对象示例 /// /// 按页取数。返回文章编号列表。 /// /// 上一页最后一笔的score,如果为空,则说明是取第一页。 /// true,用户上滑浏览下一页数据;false,用户上滑浏览最新一页数据 /// 返回key-value列表,key就是文章编号,value就是自增id(可用于lastScore) public static IDictionary GetUserPageData(string uid, int pageSize, string lastInfo, bool getPast) { long lastScore = 0; //1.解析lastInfo信息。->getPast为false,则固定取最新第一页数据,不用解析。lastInfo为空,则也不用解析,默认第一页 if (getPast && !string.IsNullOrWhiteSpace(lastInfo)) { lastScore = long.Parse(lastInfo);//外层有try..catch.. } string cacheKey = $"usr:art:{uid}"; bool isFirstPage = lastScore 0) { //如果db中也无数据,则向zset中添加一笔NoDbDataMark标识 redis.AddItemToSortedSet(cacheKey, NoDbDataMark, double.MaxValue); } } else if (items.Count == 1 && items.ContainsKey(NoDbDataMark)) { //2.2如果取到的是NoDbDataMark标识,则说明是空数据,则要Clear,返回空列表 items.Clear(); } //设置缓存有效期,要根据业务场景合理设置缓存有效期,这边以7天为例。 redis.ExpireEntryIn(cacheKey, new TimeSpan(7, 0, 0, 0)); //2.3 第一页,有多少就返回多少数据。数据如果不够一页,说明本身数据不够。 return items; } else { //3.第二页(及之后)取数 var items = GetPageDataByLastScoreFromRedis(redis, cacheKey, pageSize, lastScore); if (items.Count (string)x.aid, y => (double)y.id); } } #region 通用函数 /// /// ZSet第一页之后的取数,从lastScore开始取pageSize笔数据(第一页之后才有lastScore)。 /// 使用lua,保证原子性操作。 /// public static Dictionary GetPageDataByLastScoreFromRedis(IRedisClient redis, string zsetKey, int pageSize, double lastScore) { //ZREVRANGEBYSCORE: from lastScore to '-inf'. var luaBody = @"local sets = redis.call('ZREVRANGEBYSCORE', KEYS[1], ARGV[1], '-inf', 'WITHSCORES'); local result = {}; local index=0; local pageSize=ARGV[2]*1; local lastScore=ARGV[1]*1; for i = 1, #sets, 2 do if index>=pageSize then break; end if (lastScore>sets[i+1]*1) then table.insert(result, sets[i]); table.insert(result, sets[i+1]); index=index+1; end end return result"; //ARGV[1]:lastScore ARGV[2]:pageSize var list = redis.ExecLuaAsList(luaBody, new string[] { zsetKey }, new string[] { lastScore.ToString(), pageSize.ToString() }); var result = new Dictionary(); for (var i = 0; i < list.Count; i += 2) { result.Add(list[i], Convert.ToDouble(list[i + 1])); } return result; } /// /// 添加一项到zset缓存中。 /// /// 要添加到zset的数据项 /// 控制zset最大长度,如果为0,则不控制。 /// public static string AddItemToZset(IRedisClient redis, string zsetKey, KeyValuePair item, int maxCount = 0) { var items = new Dictionary() { { item.Key, item.Value } }; return AddItemsToZset(redis, zsetKey, items); } /// /// 添加多项到zset缓存中。 /// /// 要添加到zset的数据列表 /// 缓存zsetKey是否有设置缓存有效期。如果有设置缓存有效期,则当缓存中无数据时,可能是缓存过期;而如果缓存无有效期,缓存中无数据,就是db和缓存都无数据 /// 是否是reload情况,true重载情况;false追加 /// 控制zset最大长度,如果为0,则不控制。 /// public static string AddItemsToZset(IRedisClient redis, string zsetKey, Dictionary items, bool hasCacheExpire = true , bool isReload = false, int maxCount = 0) { //!isReload,是因为如果isReload=true情况无数据,则也要进来重载队列为无数据(即,如果之前有数据要重载为无数据) if (!isReload && items.Count 0才KeepLength。返回数值:curCount - maxCount。(可以用返回值简单算出队列当前长度curCount)。如果返回值小于等于0则说明没有触发删除操作。 * 7.maxCount为


【本文地址】


今日新闻


推荐新闻


CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3