【Redis缓存】实现对缓存数据实现排序和分页功能_redis分页排序缓存方案

link之家
链接快照平台
输入网页链接，自动生成快照
标签化管理网页链接
项目开发中往往会遇到一些查询逻辑较为复杂的报表，这些查询耗时动辄几十秒，甚至是几分钟，并且分页或排序时，往往是重新执行一遍SQL，效率低下。针对此情况，使用缓存能的解决例如排行榜和报表以及一些一致性要求不强的数据，并且对缓存数据结构的设计，可以实现对缓存数据的排序和分页功能，解决分页和排序时重新执行SQL的问题。

1）缓存SQL查询结果。
2）分页和排序通过Redis进行操作，减少数据库的查询次数。
工具：Redis 4.0.6,Jedis 2.1.0
平台：Java 1.7
数据库：MySql 5.7.17
1、生成key策略：
目的是确保唯一性，笔者采用 “方法名” + “SQL”，也可以视情况加上时间戳等条件。
2、工具类：
序列化工具。Redis不支持Java将对象直接存储到数据库中，因此需要将Java对象进行序列化，反之从Redis中取出也要反序列化。
public byte[] serialize(Object object) {
   ObjectOutputStream oos = null;
   ByteArrayOutputStream baos = null;
   try {
     baos = new ByteArrayOutputStream();
     oos = new ObjectOutputStream(baos);
     oos.writeObject(object);
     return baos.toByteArray();
    }catch (Exception e) {
     throw new CacheException(e);
 public Object unserialize(byte[] bytes) {
   if (bytes == null) {
     return null;
    ByteArrayInputStreambais = null;
   try {
     bais = new ByteArrayInputStream(bytes);
     ObjectInputStream ois = new ObjectInputStream(bais);
     return ois.readObject();
    }catch (Exception e) {
     throw new CacheException(e);
分页工具类 
public class PageData<T> {
	/** 数据集合 */
	protected List<T> result = Lists.newArrayList();
	/** 数据总数 */
	protected int totalCount = 0;
	/** 总页数 */
	protected long pageCount = 0;
	/** 每页记录 */
	protected int pageSize = 15;
	/** 初始当前页 */
	protected int pageNo = 1;
3、缓存结构设计 
缓存数据这里要分3种类型：需要分页的数据、需要分页和排序的数据和普通数据。 
3.1普通数据（既不需要分页也不需要排序） 
存储到Hash数据中 
public void putObject(final Object key,final Object value, final Integer timeout) {
        // 全局变量Hash表的ID
       finalbyte[] idBytes = id.getBytes();
       Jedisjedis = jedisPool.getResource();
       jedis.hset(idBytes,key.toString().getBytes(), SerializeUtil.serialize(value));
       if(timeout != null && jedis.ttl(idBytes) == -1) {
              jedis.expire(idBytes,timeout);
public Object getObject(final Object key) {
    Jedis jedis =jedisPool.getResource();
    return SerializeUtil.unserialize(jedis.hget(id.getBytes(),key.toString().getBytes()));
3.2有分页需求的数据 
MySql中Limit分页是将所有符合条件的数据全部查询出来，再根据Limit的参数进行截取。 
Redis中Sorted Set可以很好的完成分页的功能。将查询结果全部存储到缓存中，每次获取分页数据，都由Redis完成。 
Sorted Set API: 
    获得总记录数：zcard(key) 
    获得指定范围的结果集：zrange(key, beginIndex, endIndex) 
缓存结果图解： 
（Member记录每行结果，用逗号分隔每一列，分数记录每行结果的索引值。） 
* 保存缓存
* @param key 根据策略生成的key
* @param list 查询结果
* @param timeout 缓存过期时间
public void saveCache(final String key, List<?> list, final Integer timeout) {
	if (list.size() > 0) {
	    Jedis redis = jedisPool.getResource();
	    Pipeline pipeline = redis.pipelined();
	    // 开启事务
	    pipeline.multi();
	    for (Integer i = 0; i < list.size(); i++) {
		Object[] dataForRow = (Object[])list.get(i);
		String setValue = "";
		for (Integer j = 0; j < dataForRow.length; j++) {
		    if (dataForRow[j] == null) {
			dataForRow[j] = " ";
		    setValue += dataForRow[j].toString() + ",";
		pipeline.zadd(key, (double)i, setValue);
        // 设置过期时间
        pipeline.expire(key, timeout);
        // 提交事务
        pipeline.exec();
        pipeline.sync();
        jedisPool.returnResource(redis);
* 从缓存读取数据
* @param pageData 分页工具类
* @param key 
* @return
public PageData<Object[]> findPageDataByCache(PageData<Object[]> pageData, final String key) {
	Jedis redis = jedisPool.getResource();
	List<Object[]> cacheList = new ArrayList<Object[]>();
	if (key!= null && !"".equals(key)) {
		// 获得总记录数
		Long totalCount = redis.zcard(key);
                if (totalCount > 0) {
			// 计算分页
			Integer beginIndex = ((pageData.getPageNo() - 1) * pageData.getPageSize());
			Integer endIndex = (beginIndex + pageData.getPageSize() - 1);
			if (pageData.getTotalCount() == 0) {
				// 保存总记录数
				pageData.setTotalCount(totalCount.intValue());
			// Sorted Set返回结果集封装为LinkedHashSet
			Set<String> cacheDataForRow = redis.zrange(key, beginIndex, endIndex);
			for (String dataUnitArray : cacheDataForRow) {
				Object[] dataUnit = (Object[])dataUnitArray.split(",");
				cacheList.add(dataUnit);
			pageData.setResult(cacheList);
	jedisPool.returnResource(redis);
	return pageData;
3.3有分页和排序需求的数据 
设计原理： 
        根据存储分页数据的基础上，每一行生成一条Hash结构的数据。 
        排序需要建立列名和对应列的数据的对应关系，即获得排序字段时，能找到该列的所有数据才能进行比较。 
        Hash数据的key根据Sorted Set的分数（即每行数据的索引值） + Key生成，这样做是为了根据每行的索引值找到该set生成的所有的Hash数据。 
        根据某列进行排序，得到排序后行索引值的集合，再根据索引值取出Sorted Set中的数据。使用Redis的Sort命令实现此需求。 
现在对row0做降序处理，row0中的3条数据的值是4,1,7，因此我们希望得到的索引集合是2,1,0。 
存储每一行列名和值的Hash数据 
Sorted Set数据 
indexList存储是原Sorted Set中的索引号，即0,1,2 
Sort命令得到排序后的索引，通过Hash数据的row0列进行排序，返回indexList中的值。 
其中BY *key1->row0：
     *通配符是指从indexList中取出的数值，key1是缓存数据的key；*key1相当于生成0key1、1key1、2key1。 
    ->row0是指以“*key1”的Hash数据中取”row0”列进行比较、排序。 
    SORT indexList BY *key1->row0 DESC : 以0key1、1key1、2key1中field是row0的值组合成的新key，并按照新key中的内容进行排序，返回排序后的indexList。 
注意：若排序字段是字符串，需要加ALPHA参数。Sort命令的详细使用方法可查看官网文档。 
有了排序后的索引值集合，就可以取出排序后的结果集。 
代码部分： 
注意：根据SQL生成Key时不要将order参数写入，会导致每次查询排序时生成的Key不相同，将order作为参数交给Redis进行操作。 
	 * 生成排序索引key
	public static synchronized String createSortedListKey(String conditionId, String sortName, String sortOrder) {
		String sortedListKey = conditionId + ":" + sortName + ":" + sortOrder;
		return sortedListKey;
	 * 生成排序索引key
	public static synchronized String createSortedListKey(String conditionId, String sortName, String sortOrder) {
		String sortedListKey = conditionId + ":" + sortName + ":" + sortOrder;
		return sortedListKey;
* 保存缓存（排序）
	* @param list
	* @param rowNameList 列名集合
	* @param timeout
public void saveCacheDataWithSort(List<?> list, List<String> rowNameList , final String key, final Integer timeout) {
	if (list.size() > 0) {
		Jedis redis = jedisPool.getResource();
		Pipeline pipeline = redis.pipelined();
		// 开启事务
		pipeline.multi();
		for (Integer i = 0; i < list.size(); i++) {
			Object[] dataForRow = (Object[])list.get(i);
			String setValue = "";
			Map<String, String> resultWithRowMap = new HashMap<String, String>();
			for (Integer j = 0; j < dataForRow.length; j++) {
				if (dataForRow[j] == null) {
					dataForRow[j] = " ";
				resultWithRowMap.put(rowNameList.get(j), dataForRow[j].toString());
				setValue += dataForRow[j].toString() + ",";
			// setValueMap.put((double)i, setValue);
			pipeline.zadd(key, (double)i, setValue);
			// 将每一行数据(含列名)保存hash数据
			pipeline.hmset(i + key, resultWithRowMap);
			pipeline.expire(i + key, timeout);
		// 设置过期时间
		pipeline.expire(key, timeout);
		// 提交事务
		pipeline.exec();
		pipeline.sync();
		jedisPool.returnResource(redis);
	* 读取缓存返回pageData分页对象（含排序）
	* @param pageData
	* @param sortName
	* @param sortOrder
	* @param sortIsString 排序字段是否字符
	* @return
public PageData<Object[]> findPageDataByCacheWithSort(PageData<Object[]> pageData, final String key,	String sortName, String sortOrder, Boolean sortIsString) {
	// 生成排序索引key，缓存Sort命令排序后生成的索引集合
	String sortedListKey = CacheUtil.createSortedListKey(key, sortName, sortOrder);
 	Jedis redis = jedisPool.getResource();
	List<Object[]> cacheList = new ArrayList<Object[]>();
	if (key!= null && !"".equals(key)) {
		Integer totalCount = redis.zcard(key).intValue();
		if (totalCount > 0) {
			// 查看是是否有该排序的缓存
			byte[] bytes = redis.get(sortedListKey.getBytes());
			List<?> resultIndexList = SerializeUtil.unserializeList(bytes);
			if (resultIndexList == null || resultIndexList.isEmpty()) {
			    // 根据缓存数据的长度，创建用于存储每行数据索引的临时List
			    String uuId = UUID.randomUUID().toString().replaceAll("-", "");
			    Pipeline pipeline = redis.pipelined();
			    for (Integer i = 0; i < totalCount; i++) {
				pipeline.rpush(uuId, i.toString());
			    pipeline.sync();
			    // 排序参数对象
			    SortingParams sortingParams = new SortingParams();
			    // *是通配符,此处zset中所有分数
			    // *zsetId取出所有行数据的hash数据
			    // ->sortName指以hash数据中的sortName字段进行排序
			    sortingParams.by("*" + key + "->" + sortName);
			    if ("asc".equals(sortOrder)) {
				sortingParams.asc();
			    } else {
				sortingParams.desc();
			    if (sortIsString) {
				// 根据字符进行排序
				sortingParams.alpha();
			    // 获得排序后的索引集合
			    resultIndexList = redis.sort(uuId, sortingParams);
			    // 删除key = uuId的临时List
			    redis.ltrim(uuId, 1, 0);
			    // 将排序后的索引存入缓存,过期时间与当前报表同步
			    redis.set(sortedListKey.getBytes(), SerializeUtil.serializeList(resultIndexList));
			    redis.expire(sortedListKey, redis.ttl(key).intValue());
		    // 根据pageData属性计算分页
		    Integer beginIndex = ((pageData.getPageNo() - 1) * pageData.getPageSize());
		    Integer endIndex = (beginIndex + pageData.getPageSize());
		    if (pageData.getTotalCount() == 0) {
			// 保存总记录数
			pageData.setTotalCount(totalCount);
		    // 若取值范围大于实际结果集长度，则endIndex取实际结果集长度
		    endIndex = resultIndexList.size() < endIndex ? resultIndexList.size() : endIndex;
		    // 根据排序索引从缓存中逐条读取
		    Pipeline pipeline2 = redis.pipelined();
		    List<Response<Set<String>>> responseSetList = new ArrayList<Response<Set<String>>>();
		    for (Integer j = beginIndex; j < endIndex; j++) {
			String index = resultIndexList.get(j).toString();
			Response<Set<String>> responseSet = pipeline2.zrangeByScore(key, index, index);
			responseSetList.add(responseSet);
		    pipeline2.sync();
		    for (Response<Set<String>> response : responseSetList) {
			Set<String> cacheDataSet = response.get();
			Iterator<String> it = cacheDataSet.iterator();
			if (it.hasNext()) {
				Object[] dataUnit = (Object[])it.next().split(",");
				cacheList.add(dataUnit);
			pageData.setResult(cacheList);
    jedisPool.returnResource(redis);
    return pageData;
第一次写博客，内容很粗糙，仅供参考；欢迎各种建议、指正和交流。 
邮箱：594187062@qq.com
                                    在查询时，给获取锁添加超时时间，避免突然大量请求访问冷门数据，大量线程阻塞等待锁如果多处操作缓存和数据库，要注意加同一把锁，来保证数据的一致性在 MQ 的通知中，加锁的话，注意需要阻塞等待加锁，而不是拿不到锁就退出，因为 MQ 中的通知需要修改缓存，收到通知后是一定要修改的对于数据库中不存在的数据，在 Redis 中使用{}来进行缓存，避免缓存穿透，空缓存就可以将过期时间设置的短一些，避免大量空缓存占用缓存空间。
                                    在 Redis 中，ZSet（有序集合）是一种非常有用的数据结构，它允许我们存储一组唯一的成员，并为每个成员分配一个分数，这使得成员可以按照分数进行排序。虽然 ZSet 并没有直接提供分页功能，但我们可以通过结合使用 zRange 命令和客户端代码来实现分页效果
                                    这里写自定义目录标题欢迎使用Markdown编辑器新的改变功能快捷键合理的创建标题，有助于目录的生成如何改变文本的样式插入链接与图片如何插入一段漂亮的代码片生成一个适合你的列表创建一个表格设定内容居中、居左、居右SmartyPants创建一个自定义列表如何创建一个注脚注释也是必不可少的KaTeX数学公式新的甘特图功能，丰富你的文章UML 图表FLowchart流程图导出与导入导出导入
欢迎使用Markdown编辑器
你好！ 这是你第一次使用 Markdown编辑器 所展示的欢迎页。如果你想学习如何使用Mar
                                    作业:// 学生表  id sname cid//班级表  cid cname缓存注解一般是在service层查询所有的班级  以及班级中的所有的信息 并能缓存 到rdis里面（不要求分页）Stream流 获取第二页的数据 （ 每页数据有2条）MySQL的部分项目依赖:pom文件。
                                    Redis是一个高效的内存数据库，它支持包括、、、和等数据类型的存储，在Redis中通常根据数据的key查询其value值，Redis没有条件查询，在面对一些需要分页或排序的场景时（如评论，时间线），Redis就不太好不处理了。前段时间在项目中需要将每个主题下的用户的评论组装好写入Redis中，每个主题会有一个topicId，每一条评论会和topicId关联起来，得到大致的数据模型如下：
    topicId: 'xxxxxxxx',
    comments: [
使用缓存技术一般是在不经常增删改查的数据，我们可以使用缓存技术将第一次请求访问的数据缓存到redis中保存起来，后来的请求的时候首先查询redis数据库是否查询到这些数据，如果存在，则返回数据，如果不存在，则到mysql或其他数据库查询数据返回并保存到redis数据库中。
为什么要采用分页缓存？直接设置缓存，如果数据量大，操作增删改，更新缓存频率高效率低。分...
                                    在实现缓存排序功能之前，必须先明白这一功能的合理性。不妨思考一下，既然可以在数据库中排序，为什么还要把排序功能放在缓存中实现呢？这里简单总结了两个原因：首先，排序会增加数据库的负载，难以支撑高并发的应用；其次，在缓存中排序不会遇到表锁定的问题。Redis恰好提供了排序功能，使我们可以方便地实现缓存排序。
        Redis中用于实现排序功能的是SORT命令。该命令提供了多种参数，可以
                                    1.在实现缓存排序功能之前，必须先明白这一功能的合理性。不妨思考一下，既然可以在数据库中排序，为什么还要把排序功能放在缓存中实现呢？这里简单总结了两个原因：首先，排序会增加数据库的负载，难以支撑高并发的应用；其次，在缓存中排序不会遇到表锁定的问题。Redis恰好提供了排序功能，使我们可以方便地实现缓存排序。
应用Redis实现数据的读写，同时利用队列处理器定时将数据写入mysql。...
                                    使用Redis实现分页出发点基础架构Redis Sort功能快捷键合理的创建标题，有助于目录的生成如何改变文本的样式插入链接与图片如何插入一段漂亮的代码片生成一个适合你的列表创建一个表格设定内容居中、居左、居右SmartyPants创建一个自定义列表如何创建一个注脚注释也是必不可少的KaTeX数学公式新的甘特图功能，丰富你的文章UML 图表FLowchart流程图导出与导入导出导入
在目前...
如上图，app实现基金经理的查询，可根据年化回报率，综合得分，升序或降序排列，可下拉刷新，即实现分页效果。
生产数据24000左右，不算大，但是app查询要求响应速度，不能查询表，可以把数据放到缓存中，查询缓存。
怎么存，选择什么数据结构？
需要排序，可以选择有序集合zset
ZADD key score member [[score member] [score member] ...]
key：基金的key
sorce：回报率或综合评分
member：基金编码
除此之外...