devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

redis实现延迟任务_redis实现延时任务

作者：小蓝xlanll | 2024-02-21 18:31:07

赞

踩

redis实现延时任务

一.实现思路

1.为什么redis中使用两种数据类型，list和zset？

2.在添加zset数据的时候，为什么需要预加载？

3.延迟任务服务实现

3.1 搭建heima-leadnews-schedule模块

leadnews-schedule是一个通用的服务，单独创建模块来管理任何类型的延迟任务

①：导入资料文件夹下的heima-leadnews-schedule模块到heima-leadnews-service下，如下图所示：

 ②：添加bootstrap.yml

  1. server:
  2.   port: 51701
  3. spring:
  4.   application:
  5.     name: leadnews-schedule
  6.   cloud:
  7.     nacos:
  8.       discovery:
  9.         server-addr: 192.168.200.130:8848
  10.       config:
  11.         server-addr: 192.168.200.130:8848
  12.         file-extension: yml

③：在nacos中添加对应配置，并添加数据库及mybatis-plus的配置

  1. spring:
  2.   datasource:
  3.     driver-class-name: com.mysql.jdbc.Driver
  4.     url: jdbc:mysql://localhost:3306/leadnews_schedule?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC
  5.     username: root
  6.     password: root
  7. # 设置Mapper接口所对应的XML文件位置，如果你在Mapper接口中有自定义方法，需要进行该配置
  8. mybatis-plus:
  9.   mapper-locations: classpath*:mapper/*.xml
  10.   # 设置别名包扫描路径，通过该属性可以给包中的类注册别名
  11.   type-aliases-package: com.heima.model.schedule.pojos

4.数据库准备

taskinfo 任务表

 实体类

  1. package com.heima.model.schedule.pojos;
  2.  
  3. import com.baomidou.mybatisplus.annotation.IdType;
  4. import com.baomidou.mybatisplus.annotation.TableField;
  5. import com.baomidou.mybatisplus.annotation.TableId;
  6. import com.baomidou.mybatisplus.annotation.TableName;
  7. import lombok.Data;
  8.  
  9. import java.io.Serializable;
  10. import java.util.Date;
  11.  
  12. /**
  13.  * <p>
  14.  * 
  15.  * </p>
  16.  *
  17.  * @author itheima
  18.  */
  19. @Data
  20. @TableName("taskinfo")
  21. public class Taskinfo implements Serializable {
  22.  
  23.     private static final long serialVersionUID = 1L;
  24.  
  25.     /**
  26.      * 任务id
  27.      */
  28.     @TableId(type = IdType.ID_WORKER)
  29.     private Long taskId;
  30.  
  31.     /**
  32.      * 执行时间
  33.      */
  34.     @TableField("execute_time")
  35.     private Date executeTime;
  36.  
  37.     /**
  38.      * 参数
  39.      */
  40.     @TableField("parameters")
  41.     private byte[] parameters;
  42.  
  43.     /**
  44.      * 优先级
  45.      */
  46.     @TableField("priority")
  47.     private Integer priority;
  48.  
  49.     /**
  50.      * 任务类型
  51.      */
  52.     @TableField("task_type")
  53.     private Integer taskType;
  54.  
  55.  
  56. }

taskinfo_logs 任务日志表

实体类

  1. package com.heima.model.schedule.pojos;
  2.  
  3. import com.baomidou.mybatisplus.annotation.*;
  4. import lombok.Data;
  5.  
  6. import java.io.Serializable;
  7. import java.util.Date;
  8.  
  9. /**
  10.  * <p>
  11.  * 
  12.  * </p>
  13.  *
  14.  * @author itheima
  15.  */
  16. @Data
  17. @TableName("taskinfo_logs")
  18. public class TaskinfoLogs implements Serializable {
  19.  
  20.     private static final long serialVersionUID = 1L;
  21.  
  22.     /**
  23.      * 任务id
  24.      */
  25.     @TableId(type = IdType.ID_WORKER)
  26.     private Long taskId;
  27.  
  28.     /**
  29.      * 执行时间
  30.      */
  31.     @TableField("execute_time")
  32.     private Date executeTime;
  33.  
  34.     /**
  35.      * 参数
  36.      */
  37.     @TableField("parameters")
  38.     private byte[] parameters;
  39.  
  40.     /**
  41.      * 优先级
  42.      */
  43.     @TableField("priority")
  44.     private Integer priority;
  45.  
  46.     /**
  47.      * 任务类型
  48.      */
  49.     @TableField("task_type")
  50.     private Integer taskType;
  51.  
  52.     /**
  53.      * 版本号,用乐观锁
  54.      */
  55.     @Version
  56.     private Integer version;
  57.  
  58.     /**
  59.      * 状态 0=int(初始化) 1=EXECUTED(执行成功) 2=CANCELLED(撤销)
  60.      */
  61.     @TableField("status")
  62.     private Integer status;
  63.  
  64.  
  65. }

乐观锁支持

  1. /**
  2.      * mybatis-plus乐观锁支持
  3.      * @return
  4.      */
  5. @Bean
  6. public MybatisPlusInterceptor optimisticLockerInterceptor(){
  7.     MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
  8.     interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());
  9.     return interceptor;
  10. }

5.安装redis

①拉取镜像

docker pull redis

② 创建容器

docker run -d --name redis --restart=always -p 6379:6379 redis --requirepass "leadnews"

6.项目集成redis

① 在项目导入redis相关依赖，已经完成

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-data-redis</artifactId>
  4. </dependency>
  5. <!-- redis依赖commons-pool 这个依赖一定要添加 -->
  6. <dependency>
  7.     <groupId>org.apache.commons</groupId>
  8.     <artifactId>commons-pool2</artifactId>
  9. </dependency>

② 在heima-leadnews-schedule中集成redis,添加以下nacos配置，链接上redis

  1. spring:
  2.   redis:
  3.     host: 192.168.200.130
  4.     password: leadnews
  5.     port: 6379

③ 拷贝资料文件夹下的类：CacheService到heima-leadnews-common模块下，并添加自动配置

  1. package com.heima.schedule.config;
  2.  
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.cache.annotation.CachingConfigurerSupport;
  5. import org.springframework.dao.DataAccessException;
  6. import org.springframework.data.redis.connection.DataType;
  7. import org.springframework.data.redis.connection.RedisConnection;
  8. import org.springframework.data.redis.connection.StringRedisConnection;
  9. import org.springframework.data.redis.core.Cursor;
  10. import org.springframework.data.redis.core.RedisCallback;
  11. import org.springframework.data.redis.core.ScanOptions;
  12. import org.springframework.data.redis.core.StringRedisTemplate;
  13. import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
  14. import org.springframework.lang.Nullable;
  15. import org.springframework.stereotype.Component;
  16.  
  17. import java.io.IOException;
  18. import java.util.*;
  19. import java.util.concurrent.TimeUnit;
  20.  
  21. @Component
  22. public class CacheService extends CachingConfigurerSupport {
  23.     @Autowired
  24.     private StringRedisTemplate stringRedisTemplate;
  25.  
  26.     public StringRedisTemplate getstringRedisTemplate() {
  27.         return this.stringRedisTemplate;
  28.     }
  29.  
  30.     /** -------------------key相关操作--------------------- */
  31.  
  32.     /**
  33.      * 删除key
  34.      *
  35.      * @param key
  36.      */
  37.     public void delete(String key) {
  38.         stringRedisTemplate.delete(key);
  39.     }
  40.  
  41.     /**
  42.      * 批量删除key
  43.      *
  44.      * @param keys
  45.      */
  46.     public void delete(Collection<String> keys) {
  47.         stringRedisTemplate.delete(keys);
  48.     }
  49.  
  50.     /**
  51.      * 序列化key
  52.      *
  53.      * @param key
  54.      * @return
  55.      */
  56.     public byte[] dump(String key) {
  57.         return stringRedisTemplate.dump(key);
  58.     }
  59.  
  60.     /**
  61.      * 是否存在key
  62.      *
  63.      * @param key
  64.      * @return
  65.      */
  66.     public Boolean exists(String key) {
  67.         return stringRedisTemplate.hasKey(key);
  68.     }
  69.  
  70.     /**
  71.      * 设置过期时间
  72.      *
  73.      * @param key
  74.      * @param timeout
  75.      * @param unit
  76.      * @return
  77.      */
  78.     public Boolean expire(String key, long timeout, TimeUnit unit) {
  79.         return stringRedisTemplate.expire(key, timeout, unit);
  80.     }
  81.  
  82.     /**
  83.      * 设置过期时间
  84.      *
  85.      * @param key
  86.      * @param date
  87.      * @return
  88.      */
  89.     public Boolean expireAt(String key, Date date) {
  90.         return stringRedisTemplate.expireAt(key, date);
  91.     }
  92.  
  93.     /**
  94.      * 查找匹配的key
  95.      *
  96.      * @param pattern
  97.      * @return
  98.      */
  99.     public Set<String> keys(String pattern) {
  100.         return stringRedisTemplate.keys(pattern);
  101.     }
  102.  
  103.     /**
  104.      * 将当前数据库的 key 移动到给定的数据库 db 当中
  105.      *
  106.      * @param key
  107.      * @param dbIndex
  108.      * @return
  109.      */
  110.     public Boolean move(String key, int dbIndex) {
  111.         return stringRedisTemplate.move(key, dbIndex);
  112.     }
  113.  
  114.     /**
  115.      * 移除 key 的过期时间，key 将持久保持
  116.      *
  117.      * @param key
  118.      * @return
  119.      */
  120.     public Boolean persist(String key) {
  121.         return stringRedisTemplate.persist(key);
  122.     }
  123.  
  124.     /**
  125.      * 返回 key 的剩余的过期时间
  126.      *
  127.      * @param key
  128.      * @param unit
  129.      * @return
  130.      */
  131.     public Long getExpire(String key, TimeUnit unit) {
  132.         return stringRedisTemplate.getExpire(key, unit);
  133.     }
  134.  
  135.     /**
  136.      * 返回 key 的剩余的过期时间
  137.      *
  138.      * @param key
  139.      * @return
  140.      */
  141.     public Long getExpire(String key) {
  142.         return stringRedisTemplate.getExpire(key);
  143.     }
  144.  
  145.     /**
  146.      * 从当前数据库中随机返回一个 key
  147.      *
  148.      * @return
  149.      */
  150.     public String randomKey() {
  151.         return stringRedisTemplate.randomKey();
  152.     }
  153.  
  154.     /**
  155.      * 修改 key 的名称
  156.      *
  157.      * @param oldKey
  158.      * @param newKey
  159.      */
  160.     public void rename(String oldKey, String newKey) {
  161.         stringRedisTemplate.rename(oldKey, newKey);
  162.     }
  163.  
  164.     /**
  165.      * 仅当 newkey 不存在时，将 oldKey 改名为 newkey
  166.      *
  167.      * @param oldKey
  168.      * @param newKey
  169.      * @return
  170.      */
  171.     public Boolean renameIfAbsent(String oldKey, String newKey) {
  172.         return stringRedisTemplate.renameIfAbsent(oldKey, newKey);
  173.     }
  174.  
  175.     /**
  176.      * 返回 key 所储存的值的类型
  177.      *
  178.      * @param key
  179.      * @return
  180.      */
  181.     public DataType type(String key) {
  182.         return stringRedisTemplate.type(key);
  183.     }
  184.  
  185.     /** -------------------string相关操作--------------------- */
  186.  
  187.     /**
  188.      * 设置指定 key 的值
  189.      * @param key
  190.      * @param value
  191.      */
  192.     public void set(String key, String value) {
  193.         stringRedisTemplate.opsForValue().set(key, value);
  194.     }
  195.  
  196.     /**
  197.      * 获取指定 key 的值
  198.      * @param key
  199.      * @return
  200.      */
  201.     public String get(String key) {
  202.         return stringRedisTemplate.opsForValue().get(key);
  203.     }
  204.  
  205.     /**
  206.      * 返回 key 中字符串值的子字符
  207.      * @param key
  208.      * @param start
  209.      * @param end
  210.      * @return
  211.      */
  212.     public String getRange(String key, long start, long end) {
  213.         return stringRedisTemplate.opsForValue().get(key, start, end);
  214.     }
  215.  
  216.     /**
  217.      * 将给定 key 的值设为 value ，并返回 key 的旧值(old value)
  218.      *
  219.      * @param key
  220.      * @param value
  221.      * @return
  222.      */
  223.     public String getAndSet(String key, String value) {
  224.         return stringRedisTemplate.opsForValue().getAndSet(key, value);
  225.     }
  226.  
  227.     /**
  228.      * 对 key 所储存的字符串值，获取指定偏移量上的位(bit)
  229.      *
  230.      * @param key
  231.      * @param offset
  232.      * @return
  233.      */
  234.     public Boolean getBit(String key, long offset) {
  235.         return stringRedisTemplate.opsForValue().getBit(key, offset);
  236.     }
  237.  
  238.     /**
  239.      * 批量获取
  240.      *
  241.      * @param keys
  242.      * @return
  243.      */
  244.     public List<String> multiGet(Collection<String> keys) {
  245.         return stringRedisTemplate.opsForValue().multiGet(keys);
  246.     }
  247.  
  248.     /**
  249.      * 设置ASCII码, 字符串'a'的ASCII码是97, 转为二进制是'01100001', 此方法是将二进制第offset位值变为value
  250.      *
  251.      * @param key
  252.      * @param
  253.      * @param value
  254.      *            值,true为1, false为0
  255.      * @return
  256.      */
  257.     public boolean setBit(String key, long offset, boolean value) {
  258.         return stringRedisTemplate.opsForValue().setBit(key, offset, value);
  259.     }
  260.  
  261.     /**
  262.      * 将值 value 关联到 key ，并将 key 的过期时间设为 timeout
  263.      *
  264.      * @param key
  265.      * @param value
  266.      * @param timeout
  267.      *            过期时间
  268.      * @param unit
  269.      *            时间单位, 天:TimeUnit.DAYS 小时:TimeUnit.HOURS 分钟:TimeUnit.MINUTES
  270.      *            秒:TimeUnit.SECONDS 毫秒:TimeUnit.MILLISECONDS
  271.      */
  272.     public void setEx(String key, String value, long timeout, TimeUnit unit) {
  273.         stringRedisTemplate.opsForValue().set(key, value, timeout, unit);
  274.     }
  275.  
  276.     /**
  277.      * 只有在 key 不存在时设置 key 的值
  278.      *
  279.      * @param key
  280.      * @param value
  281.      * @return 之前已经存在返回false,不存在返回true
  282.      */
  283.     public boolean setIfAbsent(String key, String value) {
  284.         return stringRedisTemplate.opsForValue().setIfAbsent(key, value);
  285.     }
  286.  
  287.     /**
  288.      * 用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始
  289.      *
  290.      * @param key
  291.      * @param value
  292.      * @param offset
  293.      *            从指定位置开始覆写
  294.      */
  295.     public void setRange(String key, String value, long offset) {
  296.         stringRedisTemplate.opsForValue().set(key, value, offset);
  297.     }
  298.  
  299.     /**
  300.      * 获取字符串的长度
  301.      *
  302.      * @param key
  303.      * @return
  304.      */
  305.     public Long size(String key) {
  306.         return stringRedisTemplate.opsForValue().size(key);
  307.     }
  308.  
  309.     /**
  310.      * 批量添加
  311.      *
  312.      * @param maps
  313.      */
  314.     public void multiSet(Map<String, String> maps) {
  315.         stringRedisTemplate.opsForValue().multiSet(maps);
  316.     }
  317.  
  318.     /**
  319.      * 同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在
  320.      *
  321.      * @param maps
  322.      * @return 之前已经存在返回false,不存在返回true
  323.      */
  324.     public boolean multiSetIfAbsent(Map<String, String> maps) {
  325.         return stringRedisTemplate.opsForValue().multiSetIfAbsent(maps);
  326.     }
  327.  
  328.     /**
  329.      * 增加(自增长), 负数则为自减
  330.      *
  331.      * @param key
  332.      * @param
  333.      * @return
  334.      */
  335.     public Long incrBy(String key, long increment) {
  336.         return stringRedisTemplate.opsForValue().increment(key, increment);
  337.     }
  338.  
  339.     /**
  340.      *
  341.      * @param key
  342.      * @param
  343.      * @return
  344.      */
  345.     public Double incrByFloat(String key, double increment) {
  346.         return stringRedisTemplate.opsForValue().increment(key, increment);
  347.     }
  348.  
  349.     /**
  350.      * 追加到末尾
  351.      *
  352.      * @param key
  353.      * @param value
  354.      * @return
  355.      */
  356.     public Integer append(String key, String value) {
  357.         return stringRedisTemplate.opsForValue().append(key, value);
  358.     }
  359.  
  360.     /** -------------------hash相关操作------------------------- */
  361.  
  362.     /**
  363.      * 获取存储在哈希表中指定字段的值
  364.      *
  365.      * @param key
  366.      * @param field
  367.      * @return
  368.      */
  369.     public Object hGet(String key, String field) {
  370.         return stringRedisTemplate.opsForHash().get(key, field);
  371.     }
  372.  
  373.     /**
  374.      * 获取所有给定字段的值
  375.      *
  376.      * @param key
  377.      * @return
  378.      */
  379.     public Map<Object, Object> hGetAll(String key) {
  380.         return stringRedisTemplate.opsForHash().entries(key);
  381.     }
  382.  
  383.     /**
  384.      * 获取所有给定字段的值
  385.      *
  386.      * @param key
  387.      * @param fields
  388.      * @return
  389.      */
  390.     public List<Object> hMultiGet(String key, Collection<Object> fields) {
  391.         return stringRedisTemplate.opsForHash().multiGet(key, fields);
  392.     }
  393.  
  394.     public void hPut(String key, String hashKey, String value) {
  395.         stringRedisTemplate.opsForHash().put(key, hashKey, value);
  396.     }
  397.  
  398.     public void hPutAll(String key, Map<String, String> maps) {
  399.         stringRedisTemplate.opsForHash().putAll(key, maps);
  400.     }
  401.  
  402.     /**
  403.      * 仅当hashKey不存在时才设置
  404.      *
  405.      * @param key
  406.      * @param hashKey
  407.      * @param value
  408.      * @return
  409.      */
  410.     public Boolean hPutIfAbsent(String key, String hashKey, String value) {
  411.         return stringRedisTemplate.opsForHash().putIfAbsent(key, hashKey, value);
  412.     }
  413.  
  414.     /**
  415.      * 删除一个或多个哈希表字段
  416.      *
  417.      * @param key
  418.      * @param fields
  419.      * @return
  420.      */
  421.     public Long hDelete(String key, Object... fields) {
  422.         return stringRedisTemplate.opsForHash().delete(key, fields);
  423.     }
  424.  
  425.     /**
  426.      * 查看哈希表 key 中，指定的字段是否存在
  427.      *
  428.      * @param key
  429.      * @param field
  430.      * @return
  431.      */
  432.     public boolean hExists(String key, String field) {
  433.         return stringRedisTemplate.opsForHash().hasKey(key, field);
  434.     }
  435.  
  436.     /**
  437.      * 为哈希表 key 中的指定字段的整数值加上增量 increment
  438.      *
  439.      * @param key
  440.      * @param field
  441.      * @param increment
  442.      * @return
  443.      */
  444.     public Long hIncrBy(String key, Object field, long increment) {
  445.         return stringRedisTemplate.opsForHash().increment(key, field, increment);
  446.     }
  447.  
  448.     /**
  449.      * 为哈希表 key 中的指定字段的整数值加上增量 increment
  450.      *
  451.      * @param key
  452.      * @param field
  453.      * @param delta
  454.      * @return
  455.      */
  456.     public Double hIncrByFloat(String key, Object field, double delta) {
  457.         return stringRedisTemplate.opsForHash().increment(key, field, delta);
  458.     }
  459.  
  460.     /**
  461.      * 获取所有哈希表中的字段
  462.      *
  463.      * @param key
  464.      * @return
  465.      */
  466.     public Set<Object> hKeys(String key) {
  467.         return stringRedisTemplate.opsForHash().keys(key);
  468.     }
  469.  
  470.     /**
  471.      * 获取哈希表中字段的数量
  472.      *
  473.      * @param key
  474.      * @return
  475.      */
  476.     public Long hSize(String key) {
  477.         return stringRedisTemplate.opsForHash().size(key);
  478.     }
  479.  
  480.     /**
  481.      * 获取哈希表中所有值
  482.      *
  483.      * @param key
  484.      * @return
  485.      */
  486.     public List<Object> hValues(String key) {
  487.         return stringRedisTemplate.opsForHash().values(key);
  488.     }
  489.  
  490.     /**
  491.      * 迭代哈希表中的键值对
  492.      *
  493.      * @param key
  494.      * @param options
  495.      * @return
  496.      */
  497.     public Cursor<Map.Entry<Object, Object>> hScan(String key, ScanOptions options) {
  498.         return stringRedisTemplate.opsForHash().scan(key, options);
  499.     }
  500.  
  501.     /** ------------------------list相关操作---------------------------- */
  502.  
  503.     /**
  504.      * 通过索引获取列表中的元素
  505.      *
  506.      * @param key
  507.      * @param index
  508.      * @return
  509.      */
  510.     public String lIndex(String key, long index) {
  511.         return stringRedisTemplate.opsForList().index(key, index);
  512.     }
  513.  
  514.     /**
  515.      * 获取列表指定范围内的元素
  516.      *
  517.      * @param key
  518.      * @param start
  519.      *            开始位置, 0是开始位置
  520.      * @param end
  521.      *            结束位置, -1返回所有
  522.      * @return
  523.      */
  524.     public List<String> lRange(String key, long start, long end) {
  525.         return stringRedisTemplate.opsForList().range(key, start, end);
  526.     }
  527.  
  528.     /**
  529.      * 存储在list头部
  530.      *
  531.      * @param key
  532.      * @param value
  533.      * @return
  534.      */
  535.     public Long lLeftPush(String key, String value) {
  536.         return stringRedisTemplate.opsForList().leftPush(key, value);
  537.     }
  538.  
  539.     /**
  540.      *
  541.      * @param key
  542.      * @param value
  543.      * @return
  544.      */
  545.     public Long lLeftPushAll(String key, String... value) {
  546.         return stringRedisTemplate.opsForList().leftPushAll(key, value);
  547.     }
  548.  
  549.     /**
  550.      *
  551.      * @param key
  552.      * @param value
  553.      * @return
  554.      */
  555.     public Long lLeftPushAll(String key, Collection<String> value) {
  556.         return stringRedisTemplate.opsForList().leftPushAll(key, value);
  557.     }
  558.  
  559.     /**
  560.      * 当list存在的时候才加入
  561.      *
  562.      * @param key
  563.      * @param value
  564.      * @return
  565.      */
  566.     public Long lLeftPushIfPresent(String key, String value) {
  567.         return stringRedisTemplate.opsForList().leftPushIfPresent(key, value);
  568.     }
  569.  
  570.     /**
  571.      * 如果pivot存在,再pivot前面添加
  572.      *
  573.      * @param key
  574.      * @param pivot
  575.      * @param value
  576.      * @return
  577.      */
  578.     public Long lLeftPush(String key, String pivot, String value) {
  579.         return stringRedisTemplate.opsForList().leftPush(key, pivot, value);
  580.     }
  581.  
  582.     /**
  583.      *
  584.      * @param key
  585.      * @param value
  586.      * @return
  587.      */
  588.     public Long lRightPush(String key, String value) {
  589.         return stringRedisTemplate.opsForList().rightPush(key, value);
  590.     }
  591.  
  592.     /**
  593.      *
  594.      * @param key
  595.      * @param value
  596.      * @return
  597.      */
  598.     public Long lRightPushAll(String key, String... value) {
  599.         return stringRedisTemplate.opsForList().rightPushAll(key, value);
  600.     }
  601.  
  602.     /**
  603.      *
  604.      * @param key
  605.      * @param value
  606.      * @return
  607.      */
  608.     public Long lRightPushAll(String key, Collection<String> value) {
  609.         return stringRedisTemplate.opsForList().rightPushAll(key, value);
  610.     }
  611.  
  612.     /**
  613.      * 为已存在的列表添加值
  614.      *
  615.      * @param key
  616.      * @param value
  617.      * @return
  618.      */
  619.     public Long lRightPushIfPresent(String key, String value) {
  620.         return stringRedisTemplate.opsForList().rightPushIfPresent(key, value);
  621.     }
  622.  
  623.     /**
  624.      * 在pivot元素的右边添加值
  625.      *
  626.      * @param key
  627.      * @param pivot
  628.      * @param value
  629.      * @return
  630.      */
  631.     public Long lRightPush(String key, String pivot, String value) {
  632.         return stringRedisTemplate.opsForList().rightPush(key, pivot, value);
  633.     }
  634.  
  635.     /**
  636.      * 通过索引设置列表元素的值
  637.      *
  638.      * @param key
  639.      * @param index
  640.      *            位置
  641.      * @param value
  642.      */
  643.     public void lSet(String key, long index, String value) {
  644.         stringRedisTemplate.opsForList().set(key, index, value);
  645.     }
  646.  
  647.     /**
  648.      * 移出并获取列表的第一个元素
  649.      *
  650.      * @param key
  651.      * @return 删除的元素
  652.      */
  653.     public String lLeftPop(String key) {
  654.         return stringRedisTemplate.opsForList().leftPop(key);
  655.     }
  656.  
  657.     /**
  658.      * 移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止
  659.      *
  660.      * @param key
  661.      * @param timeout
  662.      *            等待时间
  663.      * @param unit
  664.      *            时间单位
  665.      * @return
  666.      */
  667.     public String lBLeftPop(String key, long timeout, TimeUnit unit) {
  668.         return stringRedisTemplate.opsForList().leftPop(key, timeout, unit);
  669.     }
  670.  
  671.     /**
  672.      * 移除并获取列表最后一个元素
  673.      *
  674.      * @param key
  675.      * @return 删除的元素
  676.      */
  677.     public String lRightPop(String key) {
  678.         return stringRedisTemplate.opsForList().rightPop(key);
  679.     }
  680.  
  681.     /**
  682.      * 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止
  683.      *
  684.      * @param key
  685.      * @param timeout
  686.      *            等待时间
  687.      * @param unit
  688.      *            时间单位
  689.      * @return
  690.      */
  691.     public String lBRightPop(String key, long timeout, TimeUnit unit) {
  692.         return stringRedisTemplate.opsForList().rightPop(key, timeout, unit);
  693.     }
  694.  
  695.     /**
  696.      * 移除列表的最后一个元素，并将该元素添加到另一个列表并返回
  697.      *
  698.      * @param sourceKey
  699.      * @param destinationKey
  700.      * @return
  701.      */
  702.     public String lRightPopAndLeftPush(String sourceKey, String destinationKey) {
  703.         return stringRedisTemplate.opsForList().rightPopAndLeftPush(sourceKey,
  704.                 destinationKey);
  705.     }
  706.  
  707.     /**
  708.      * 从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止
  709.      *
  710.      * @param sourceKey
  711.      * @param destinationKey
  712.      * @param timeout
  713.      * @param unit
  714.      * @return
  715.      */
  716.     public String lBRightPopAndLeftPush(String sourceKey, String destinationKey,
  717.                                         long timeout, TimeUnit unit) {
  718.         return stringRedisTemplate.opsForList().rightPopAndLeftPush(sourceKey,
  719.                 destinationKey, timeout, unit);
  720.     }
  721.     
  722.     /**
  723.      * 删除集合中值等于value得元素
  724.      *
  725.      * @param key
  726.      * @param index
  727.      *            index=0, 删除所有值等于value的元素; index>0, 从头部开始删除第一个值等于value的元素;
  728.      *            index<0, 从尾部开始删除第一个值等于value的元素;
  729.      * @param value
  730.      * @return
  731.      */
  732.     public Long lRemove(String key, long index, String value) {
  733.         return stringRedisTemplate.opsForList().remove(key, index, value);
  734.     }
  735.  
  736.     /**
  737.      * 裁剪list
  738.      *
  739.      * @param key
  740.      * @param start
  741.      * @param end
  742.      */
  743.     public void lTrim(String key, long start, long end) {
  744.         stringRedisTemplate.opsForList().trim(key, start, end);
  745.     }
  746.  
  747.     /**
  748.      * 获取列表长度
  749.      *
  750.      * @param key
  751.      * @return
  752.      */
  753.     public Long lLen(String key) {
  754.         return stringRedisTemplate.opsForList().size(key);
  755.     }
  756.  
  757.  
  758.     /** --------------------set相关操作-------------------------- */
  759.  
  760.     /**
  761.      * set添加元素
  762.      *
  763.      * @param key
  764.      * @param values
  765.      * @return
  766.      */
  767.     public Long sAdd(String key, String... values) {
  768.         return stringRedisTemplate.opsForSet().add(key, values);
  769.     }
  770.  
  771.     /**
  772.      * set移除元素
  773.      *
  774.      * @param key
  775.      * @param values
  776.      * @return
  777.      */
  778.     public Long sRemove(String key, Object... values) {
  779.         return stringRedisTemplate.opsForSet().remove(key, values);
  780.     }
  781.  
  782.     /**
  783.      * 移除并返回集合的一个随机元素
  784.      *
  785.      * @param key
  786.      * @return
  787.      */
  788.     public String sPop(String key) {
  789.         return stringRedisTemplate.opsForSet().pop(key);
  790.     }
  791.  
  792.     /**
  793.      * 将元素value从一个集合移到另一个集合
  794.      *
  795.      * @param key
  796.      * @param value
  797.      * @param destKey
  798.      * @return
  799.      */
  800.     public Boolean sMove(String key, String value, String destKey) {
  801.         return stringRedisTemplate.opsForSet().move(key, value, destKey);
  802.     }
  803.  
  804.     /**
  805.      * 获取集合的大小
  806.      *
  807.      * @param key
  808.      * @return
  809.      */
  810.     public Long sSize(String key) {
  811.         return stringRedisTemplate.opsForSet().size(key);
  812.     }
  813.  
  814.     /**
  815.      * 判断集合是否包含value
  816.      *
  817.      * @param key
  818.      * @param value
  819.      * @return
  820.      */
  821.     public Boolean sIsMember(String key, Object value) {
  822.         return stringRedisTemplate.opsForSet().isMember(key, value);
  823.     }
  824.  
  825.     /**
  826.      * 获取两个集合的交集
  827.      *
  828.      * @param key
  829.      * @param otherKey
  830.      * @return
  831.      */
  832.     public Set<String> sIntersect(String key, String otherKey) {
  833.         return stringRedisTemplate.opsForSet().intersect(key, otherKey);
  834.     }
  835.  
  836.     /**
  837.      * 获取key集合与多个集合的交集
  838.      *
  839.      * @param key
  840.      * @param otherKeys
  841.      * @return
  842.      */
  843.     public Set<String> sIntersect(String key, Collection<String> otherKeys) {
  844.         return stringRedisTemplate.opsForSet().intersect(key, otherKeys);
  845.     }
  846.  
  847.     /**
  848.      * key集合与otherKey集合的交集存储到destKey集合中
  849.      *
  850.      * @param key
  851.      * @param otherKey
  852.      * @param destKey
  853.      * @return
  854.      */
  855.     public Long sIntersectAndStore(String key, String otherKey, String destKey) {
  856.         return stringRedisTemplate.opsForSet().intersectAndStore(key, otherKey,
  857.                 destKey);
  858.     }
  859.  
  860.     /**
  861.      * key集合与多个集合的交集存储到destKey集合中
  862.      *
  863.      * @param key
  864.      * @param otherKeys
  865.      * @param destKey
  866.      * @return
  867.      */
  868.     public Long sIntersectAndStore(String key, Collection<String> otherKeys,
  869.                                    String destKey) {
  870.         return stringRedisTemplate.opsForSet().intersectAndStore(key, otherKeys,
  871.                 destKey);
  872.     }
  873.  
  874.     /**
  875.      * 获取两个集合的并集
  876.      *
  877.      * @param key
  878.      * @param otherKeys
  879.      * @return
  880.      */
  881.     public Set<String> sUnion(String key, String otherKeys) {
  882.         return stringRedisTemplate.opsForSet().union(key, otherKeys);
  883.     }
  884.  
  885.     /**
  886.      * 获取key集合与多个集合的并集
  887.      *
  888.      * @param key
  889.      * @param otherKeys
  890.      * @return
  891.      */
  892.     public Set<String> sUnion(String key, Collection<String> otherKeys) {
  893.         return stringRedisTemplate.opsForSet().union(key, otherKeys);
  894.     }
  895.  
  896.     /**
  897.      * key集合与otherKey集合的并集存储到destKey中
  898.      *
  899.      * @param key
  900.      * @param otherKey
  901.      * @param destKey
  902.      * @return
  903.      */
  904.     public Long sUnionAndStore(String key, String otherKey, String destKey) {
  905.         return stringRedisTemplate.opsForSet().unionAndStore(key, otherKey, destKey);
  906.     }
  907.  
  908.     /**
  909.      * key集合与多个集合的并集存储到destKey中
  910.      *
  911.      * @param key
  912.      * @param otherKeys
  913.      * @param destKey
  914.      * @return
  915.      */
  916.     public Long sUnionAndStore(String key, Collection<String> otherKeys,
  917.                                String destKey) {
  918.         return stringRedisTemplate.opsForSet().unionAndStore(key, otherKeys, destKey);
  919.     }
  920.  
  921.     /**
  922.      * 获取两个集合的差集
  923.      *
  924.      * @param key
  925.      * @param otherKey
  926.      * @return
  927.      */
  928.     public Set<String> sDifference(String key, String otherKey) {
  929.         return stringRedisTemplate.opsForSet().difference(key, otherKey);
  930.     }
  931.  
  932.     /**
  933.      * 获取key集合与多个集合的差集
  934.      *
  935.      * @param key
  936.      * @param otherKeys
  937.      * @return
  938.      */
  939.     public Set<String> sDifference(String key, Collection<String> otherKeys) {
  940.         return stringRedisTemplate.opsForSet().difference(key, otherKeys);
  941.     }
  942.  
  943.     /**
  944.      * key集合与otherKey集合的差集存储到destKey中
  945.      *
  946.      * @param key
  947.      * @param otherKey
  948.      * @param destKey
  949.      * @return
  950.      */
  951.     public Long sDifference(String key, String otherKey, String destKey) {
  952.         return stringRedisTemplate.opsForSet().differenceAndStore(key, otherKey,
  953.                 destKey);
  954.     }
  955.  
  956.     /**
  957.      * key集合与多个集合的差集存储到destKey中
  958.      *
  959.      * @param key
  960.      * @param otherKeys
  961.      * @param destKey
  962.      * @return
  963.      */
  964.     public Long sDifference(String key, Collection<String> otherKeys,
  965.                             String destKey) {
  966.         return stringRedisTemplate.opsForSet().differenceAndStore(key, otherKeys,
  967.                 destKey);
  968.     }
  969.  
  970.     /**
  971.      * 获取集合所有元素
  972.      *
  973.      * @param key
  974.      * @param
  975.      * @param
  976.      * @return
  977.      */
  978.     public Set<String> setMembers(String key) {
  979.         return stringRedisTemplate.opsForSet().members(key);
  980.     }
  981.  
  982.     /**
  983.      * 随机获取集合中的一个元素
  984.      *
  985.      * @param key
  986.      * @return
  987.      */
  988.     public String sRandomMember(String key) {
  989.         return stringRedisTemplate.opsForSet().randomMember(key);
  990.     }
  991.  
  992.     /**
  993.      * 随机获取集合中count个元素
  994.      *
  995.      * @param key
  996.      * @param count
  997.      * @return
  998.      */
  999.     public List<String> sRandomMembers(String key, long count) {
  1000.         return stringRedisTemplate.opsForSet().randomMembers(key, count);
  1001.     }
  1002.  
  1003.     /**
  1004.      * 随机获取集合中count个元素并且去除重复的
  1005.      *
  1006.      * @param key
  1007.      * @param count
  1008.      * @return
  1009.      */
  1010.     public Set<String> sDistinctRandomMembers(String key, long count) {
  1011.         return stringRedisTemplate.opsForSet().distinctRandomMembers(key, count);
  1012.     }
  1013.  
  1014.     /**
  1015.      *
  1016.      * @param key
  1017.      * @param options
  1018.      * @return
  1019.      */
  1020.     public Cursor<String> sScan(String key, ScanOptions options) {
  1021.         return stringRedisTemplate.opsForSet().scan(key, options);
  1022.     }
  1023.  
  1024.     /**------------------zSet相关操作--------------------------------*/
  1025.  
  1026.     /**
  1027.      * 添加元素,有序集合是按照元素的score值由小到大排列
  1028.      *
  1029.      * @param key
  1030.      * @param value
  1031.      * @param score
  1032.      * @return
  1033.      */
  1034.     public Boolean zAdd(String key, String value, double score) {
  1035.         return stringRedisTemplate.opsForZSet().add(key, value, score);
  1036.     }
  1037.  
  1038.     /**
  1039.      *
  1040.      * @param key
  1041.      * @param values
  1042.      * @return
  1043.      */
  1044.     public Long zAdd(String key, Set<TypedTuple<String>> values) {
  1045.         return stringRedisTemplate.opsForZSet().add(key, values);
  1046.     }
  1047.  
  1048.     /**
  1049.      *
  1050.      * @param key
  1051.      * @param values
  1052.      * @return
  1053.      */
  1054.     public Long zRemove(String key, Object... values) {
  1055.         return stringRedisTemplate.opsForZSet().remove(key, values);
  1056.     }
  1057.  
  1058.     public Long zRemove(String key, Collection<String> values) {
  1059.         if(values!=null&&!values.isEmpty()){
  1060.             Object[] objs = values.toArray(new Object[values.size()]);
  1061.             return stringRedisTemplate.opsForZSet().remove(key, objs);
  1062.         }
  1063.        return 0L;
  1064.     }
  1065.  
  1066.     /**
  1067.      * 增加元素的score值，并返回增加后的值
  1068.      *
  1069.      * @param key
  1070.      * @param value
  1071.      * @param delta
  1072.      * @return
  1073.      */
  1074.     public Double zIncrementScore(String key, String value, double delta) {
  1075.         return stringRedisTemplate.opsForZSet().incrementScore(key, value, delta);
  1076.     }
  1077.  
  1078.     /**
  1079.      * 返回元素在集合的排名,有序集合是按照元素的score值由小到大排列
  1080.      *
  1081.      * @param key
  1082.      * @param value
  1083.      * @return 0表示第一位
  1084.      */
  1085.     public Long zRank(String key, Object value) {
  1086.         return stringRedisTemplate.opsForZSet().rank(key, value);
  1087.     }
  1088.  
  1089.     /**
  1090.      * 返回元素在集合的排名,按元素的score值由大到小排列
  1091.      *
  1092.      * @param key
  1093.      * @param value
  1094.      * @return
  1095.      */
  1096.     public Long zReverseRank(String key, Object value) {
  1097.         return stringRedisTemplate.opsForZSet().reverseRank(key, value);
  1098.     }
  1099.  
  1100.     /**
  1101.      * 获取集合的元素, 从小到大排序
  1102.      *
  1103.      * @param key
  1104.      * @param start
  1105.      *            开始位置
  1106.      * @param end
  1107.      *            结束位置, -1查询所有
  1108.      * @return
  1109.      */
  1110.     public Set<String> zRange(String key, long start, long end) {
  1111.         return stringRedisTemplate.opsForZSet().range(key, start, end);
  1112.     }
  1113.     
  1114.     /**
  1115.      * 获取zset集合的所有元素, 从小到大排序
  1116.      *
  1117.      */
  1118.     public Set<String> zRangeAll(String key) {
  1119.         return zRange(key,0,-1);
  1120.     }
  1121.  
  1122.     /**
  1123.      * 获取集合元素, 并且把score值也获取
  1124.      *
  1125.      * @param key
  1126.      * @param start
  1127.      * @param end
  1128.      * @return
  1129.      */
  1130.     public Set<TypedTuple<String>> zRangeWithScores(String key, long start,
  1131.                                                     long end) {
  1132.         return stringRedisTemplate.opsForZSet().rangeWithScores(key, start, end);
  1133.     }
  1134.  
  1135.     /**
  1136.      * 根据Score值查询集合元素
  1137.      *
  1138.      * @param key
  1139.      * @param min
  1140.      *            最小值
  1141.      * @param max
  1142.      *            最大值
  1143.      * @return
  1144.      */
  1145.     public Set<String> zRangeByScore(String key, double min, double max) {
  1146.         return stringRedisTemplate.opsForZSet().rangeByScore(key, min, max);
  1147.     }
  1148.  
  1149.  
  1150.     /**
  1151.      * 根据Score值查询集合元素, 从小到大排序
  1152.      *
  1153.      * @param key
  1154.      * @param min
  1155.      *            最小值
  1156.      * @param max
  1157.      *            最大值
  1158.      * @return
  1159.      */
  1160.     public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,
  1161.                                                            double min, double max) {
  1162.         return stringRedisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max);
  1163.     }
  1164.  
  1165.     /**
  1166.      *
  1167.      * @param key
  1168.      * @param min
  1169.      * @param max
  1170.      * @param start
  1171.      * @param end
  1172.      * @return
  1173.      */
  1174.     public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,
  1175.                                                            double min, double max, long start, long end) {
  1176.         return stringRedisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max,
  1177.                 start, end);
  1178.     }
  1179.  
  1180.     /**
  1181.      * 获取集合的元素, 从大到小排序
  1182.      *
  1183.      * @param key
  1184.      * @param start
  1185.      * @param end
  1186.      * @return
  1187.      */
  1188.     public Set<String> zReverseRange(String key, long start, long end) {
  1189.         return stringRedisTemplate.opsForZSet().reverseRange(key, start, end);
  1190.  
  1191.     }
  1192.  
  1193.     public Set<String> zReverseRangeByScore(String key, long min, long max) {
  1194.         return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max);
  1195.  
  1196.     }
  1197.  
  1198.     /**
  1199.      * 获取集合的元素, 从大到小排序, 并返回score值
  1200.      *
  1201.      * @param key
  1202.      * @param start
  1203.      * @param end
  1204.      * @return
  1205.      */
  1206.     public Set<TypedTuple<String>> zReverseRangeWithScores(String key,
  1207.                                                            long start, long end) {
  1208.         return stringRedisTemplate.opsForZSet().reverseRangeWithScores(key, start,
  1209.                 end);
  1210.     }
  1211.  
  1212.     /**
  1213.      * 根据Score值查询集合元素, 从大到小排序
  1214.      *
  1215.      * @param key
  1216.      * @param min
  1217.      * @param max
  1218.      * @return
  1219.      */
  1220.     public Set<String> zReverseRangeByScore(String key, double min,
  1221.                                             double max) {
  1222.         return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max);
  1223.     }
  1224.  
  1225.     /**
  1226.      * 根据Score值查询集合元素, 从大到小排序
  1227.      *
  1228.      * @param key
  1229.      * @param min
  1230.      * @param max
  1231.      * @return
  1232.      */
  1233.     public Set<TypedTuple<String>> zReverseRangeByScoreWithScores(
  1234.             String key, double min, double max) {
  1235.         return stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key,
  1236.                 min, max);
  1237.     }
  1238.  
  1239.     /**
  1240.      *
  1241.      * @param key
  1242.      * @param min
  1243.      * @param max
  1244.      * @param start
  1245.      * @param end
  1246.      * @return
  1247.      */
  1248.     public Set<String> zReverseRangeByScore(String key, double min,
  1249.                                             double max, long start, long end) {
  1250.         return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max,
  1251.                 start, end);
  1252.     }
  1253.  
  1254.     /**
  1255.      * 根据score值获取集合元素数量
  1256.      *
  1257.      * @param key
  1258.      * @param min
  1259.      * @param max
  1260.      * @return
  1261.      */
  1262.     public Long zCount(String key, double min, double max) {
  1263.         return stringRedisTemplate.opsForZSet().count(key, min, max);
  1264.     }
  1265.  
  1266.     /**
  1267.      * 获取集合大小
  1268.      *
  1269.      * @param key
  1270.      * @return
  1271.      */
  1272.     public Long zSize(String key) {
  1273.         return stringRedisTemplate.opsForZSet().size(key);
  1274.     }
  1275.  
  1276.     /**
  1277.      * 获取集合大小
  1278.      *
  1279.      * @param key
  1280.      * @return
  1281.      */
  1282.     public Long zZCard(String key) {
  1283.         return stringRedisTemplate.opsForZSet().zCard(key);
  1284.     }
  1285.  
  1286.     /**
  1287.      * 获取集合中value元素的score值
  1288.      *
  1289.      * @param key
  1290.      * @param value
  1291.      * @return
  1292.      */
  1293.     public Double zScore(String key, Object value) {
  1294.         return stringRedisTemplate.opsForZSet().score(key, value);
  1295.     }
  1296.  
  1297.     /**
  1298.      * 移除指定索引位置的成员
  1299.      *
  1300.      * @param key
  1301.      * @param start
  1302.      * @param end
  1303.      * @return
  1304.      */
  1305.     public Long zRemoveRange(String key, long start, long end) {
  1306.         return stringRedisTemplate.opsForZSet().removeRange(key, start, end);
  1307.     }
  1308.  
  1309.     /**
  1310.      * 根据指定的score值的范围来移除成员
  1311.      *
  1312.      * @param key
  1313.      * @param min
  1314.      * @param max
  1315.      * @return
  1316.      */
  1317.     public Long zRemoveRangeByScore(String key, double min, double max) {
  1318.         return stringRedisTemplate.opsForZSet().removeRangeByScore(key, min, max);
  1319.     }
  1320.  
  1321.     /**
  1322.      * 获取key和otherKey的并集并存储在destKey中
  1323.      *
  1324.      * @param key
  1325.      * @param otherKey
  1326.      * @param destKey
  1327.      * @return
  1328.      */
  1329.     public Long zUnionAndStore(String key, String otherKey, String destKey) {
  1330.         return stringRedisTemplate.opsForZSet().unionAndStore(key, otherKey, destKey);
  1331.     }
  1332.  
  1333.     /**
  1334.      *
  1335.      * @param key
  1336.      * @param otherKeys
  1337.      * @param destKey
  1338.      * @return
  1339.      */
  1340.     public Long zUnionAndStore(String key, Collection<String> otherKeys,
  1341.                                String destKey) {
  1342.         return stringRedisTemplate.opsForZSet()
  1343.                 .unionAndStore(key, otherKeys, destKey);
  1344.     }
  1345.  
  1346.     /**
  1347.      * 交集
  1348.      *
  1349.      * @param key
  1350.      * @param otherKey
  1351.      * @param destKey
  1352.      * @return
  1353.      */
  1354.     public Long zIntersectAndStore(String key, String otherKey,
  1355.                                    String destKey) {
  1356.         return stringRedisTemplate.opsForZSet().intersectAndStore(key, otherKey,
  1357.                 destKey);
  1358.     }
  1359.  
  1360.     /**
  1361.      * 交集
  1362.      *
  1363.      * @param key
  1364.      * @param otherKeys
  1365.      * @param destKey
  1366.      * @return
  1367.      */
  1368.     public Long zIntersectAndStore(String key, Collection<String> otherKeys,
  1369.                                    String destKey) {
  1370.         return stringRedisTemplate.opsForZSet().intersectAndStore(key, otherKeys,
  1371.                 destKey);
  1372.     }
  1373.  
  1374.     /**
  1375.      *
  1376.      * @param key
  1377.      * @param options
  1378.      * @return
  1379.      */
  1380.     public Cursor<TypedTuple<String>> zScan(String key, ScanOptions options) {
  1381.         return stringRedisTemplate.opsForZSet().scan(key, options);
  1382.     }
  1383.  
  1384.     /**
  1385.      * 扫描主键，建议使用
  1386.      * @param patten
  1387.      * @return
  1388.      */
  1389.     public Set<String> scan(String patten){
  1390.         Set<String> keys = stringRedisTemplate.execute((RedisCallback<Set<String>>) connection -> {
  1391.             Set<String> result = new HashSet<>();
  1392.             try (Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder()
  1393.                     .match(patten).count(10000).build())) {
  1394.                 while (cursor.hasNext()) {
  1395.                     result.add(new String(cursor.next()));
  1396.                 }
  1397.             } catch (IOException e) {
  1398.                 e.printStackTrace();
  1399.             }
  1400.             return result;
  1401.         });
  1402.         return  keys;
  1403.     }
  1404.     
  1405.     /**
  1406.      * 管道技术，提高性能
  1407.      * @param type
  1408.      * @param values
  1409.      * @return
  1410.      */
  1411.     public List<Object> lRightPushPipeline(String type,Collection<String> values){
  1412.         List<Object> results = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
  1413.                     public Object doInRedis(RedisConnection connection) throws DataAccessException {
  1414.                         StringRedisConnection stringRedisConn = (StringRedisConnection)connection;
  1415.                         //集合转换数组
  1416.                         String[] strings = values.toArray(new String[values.size()]);
  1417.                         //直接批量发送
  1418.                         stringRedisConn.rPush(type, strings);
  1419.                         return null;
  1420.                     }
  1421.                 });
  1422.         return results;
  1423.     }
  1424.  
  1425.     public List<Object> refreshWithPipeline(String future_key,String topic_key,Collection<String> values){
  1426.  
  1427.         List<Object> objects = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {
  1428.             @Nullable
  1429.             @Override
  1430.             public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
  1431.                 StringRedisConnection stringRedisConnection = (StringRedisConnection)redisConnection;
  1432.                 String[] strings = values.toArray(new String[values.size()]);
  1433.                 stringRedisConnection.rPush(topic_key,strings);
  1434.                 stringRedisConnection.zRem(future_key,strings);
  1435.                 return null;
  1436.             }
  1437.         });
  1438.         return objects;
  1439.     }
  1440.  
  1441. /**
  1442.  * 加锁
  1443.  *
  1444.  * @param name
  1445.  * @param expire
  1446.  * @return
  1447.  */
  1448. public String tryLock(String name, long expire) {
  1449.     name = name + "_lock";
  1450.     String token = UUID.randomUUID().toString();
  1451.     RedisConnectionFactory factory = stringRedisTemplate.getConnectionFactory();
  1452.     RedisConnection conn = factory.getConnection();
  1453.     try {
  1454.  
  1455.         //参考redis命令：
  1456.         //set key value [EX seconds] [PX milliseconds] [NX|XX]
  1457.         Boolean result = conn.set(
  1458.                 name.getBytes(), //锁的名称
  1459.                 token.getBytes(),
  1460.                 Expiration.from(expire, TimeUnit.MILLISECONDS),//过期时间,自己设置30s,60s过期
  1461.                 RedisStringCommands.SetOption.SET_IF_ABSENT //NX
  1462.         );
  1463.         if (result != null && result)
  1464.             return token;
  1465.     } finally {
  1466.         RedisConnectionUtils.releaseConnection(conn, factory,false);
  1467.     }
  1468.     return null;
  1469. }
  1470.  
  1471. }

④：测试

  1. package com.heima.schedule.test;
  2.  
  3.  
  4. import com.heima.common.redis.CacheService;
  5. import com.heima.schedule.ScheduleApplication;
  6. import org.junit.Test;
  7. import org.junit.runner.RunWith;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.boot.test.context.SpringBootTest;
  10. import org.springframework.test.context.junit4.SpringRunner;
  11.  
  12. import java.util.Set;
  13.  
  14.  
  15. @SpringBootTest(classes = ScheduleApplication.class)
  16. @RunWith(SpringRunner.class)
  17. public class RedisTest {
  18.  
  19.     @Autowired
  20.     private CacheService cacheService;
  21.  
  22.     @Test
  23.     public void testList(){
  24.  
  25.         //在list的左边添加元素
  26. //        cacheService.lLeftPush("list_001","hello,redis");
  27.  
  28.         //在list的右边获取元素，并删除
  29.         String list_001 = cacheService.lRightPop("list_001");
  30.         System.out.println(list_001);
  31.     }
  32.  
  33.     @Test
  34.     public void testZset(){
  35.         //添加数据到zset中  分值
  36.         /*cacheService.zAdd("zset_key_001","hello zset 001",1000);
  37.         cacheService.zAdd("zset_key_001","hello zset 002",8888);
  38.         cacheService.zAdd("zset_key_001","hello zset 003",7777);
  39.         cacheService.zAdd("zset_key_001","hello zset 004",999999);*/
  40.  
  41.         //按照分值获取数据
  42.         Set<String> zset_key_001 = cacheService.zRangeByScore("zset_key_001", 0, 8888);
  43.         System.out.println(zset_key_001);
  44.  
  45.  
  46.     }
  47. }

二.redis延迟任务实现

1.添加taskinfoMapper

  1. package com.heima.schedule.mapper;
  2.  
  3. import com.baomidou.mybatisplus.core.mapper.BaseMapper;
  4. import com.heima.model.schedule.pojos.Taskinfo;
  5. import org.apache.ibatis.annotations.Mapper;
  6. import org.apache.ibatis.annotations.Param;
  7. import org.apache.ibatis.annotations.Select;
  8.  
  9. import java.util.Date;
  10. import java.util.List;
  11.  
  12. /**
  13.  * <p>
  14.  *  Mapper 接口
  15.  * </p>
  16.  *
  17.  * @author itheima
  18.  */
  19. @Mapper
  20. public interface TaskinfoMapper extends BaseMapper<Taskinfo> {
  21.  
  22.     public List<Taskinfo> queryFutureTime(@Param("taskType")int type, @Param("priority")int priority, @Param("future")Date future);
  23. }

 以及taskinfoMapper.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  3. <mapper namespace="com.heima.schedule.mapper.TaskinfoMapper">
  4.  
  5.     <select id="queryFutureTime" resultType="com.heima.model.schedule.pojos.Taskinfo">
  6.         select *
  7.         from taskinfo
  8.         where task_type = #{taskType}
  9.           and priority = #{priority}
  10.           and execute_time <![CDATA[<]]> #{future,javaType=java.util.Date}
  11.     </select>
  12.  
  13. </mapper>

2.添加taskinfoLogsMapper

  1. package com.heima.schedule.mapper;
  2.  
  3. import com.baomidou.mybatisplus.core.mapper.BaseMapper;
  4. import com.heima.model.schedule.pojos.TaskinfoLogs;
  5. import org.apache.ibatis.annotations.Mapper;
  6.  
  7. /**
  8.  * <p>
  9.  *  Mapper 接口
  10.  * </p>
  11.  *
  12.  * @author itheima
  13.  */
  14. @Mapper
  15. public interface TaskinfoLogsMapper extends BaseMapper<TaskinfoLogs> {
  16.  
  17. }

3.创建task类,用于创建添加任务参数

  1. package com.heima.model.schedule.dtos;
  2.  
  3. import lombok.Data;
  4.  
  5. import java.io.Serializable;
  6.  
  7. @Data
  8. public class Task implements Serializable {
  9.  
  10.     /**
  11.      * 任务id
  12.      */
  13.     private Long taskId;
  14.     /**
  15.      * 类型
  16.      */
  17.     private Integer taskType;
  18.  
  19.     /**
  20.      * 优先级
  21.      */
  22.     private Integer priority;
  23.  
  24.     /**
  25.      * 执行id
  26.      */
  27.     private long executeTime;
  28.  
  29.     /**
  30.      * task参数
  31.      */
  32.     private byte[] parameters;
  33.     
  34. }

4.创建taskService

  1. package com.heima.schedule.service;
  2.  
  3. import com.heima.model.schedule.dtos.Task;
  4.  
  5. /**
  6.  * 对外访问接口
  7.  */
  8. public interface TaskService {
  9.  
  10.     /**
  11.      * 添加任务
  12.      * @param task   任务对象
  13.      * @return       任务id
  14.      */
  15.     public long addTask(Task task) ;
  16.  
  17. }

实现:

  1. package com.heima.schedule.service.impl;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.heima.common.constants.ScheduleConstants;
  5. import com.heima.common.redis.CacheService;
  6. import com.heima.model.schedule.dtos.Task;
  7. import com.heima.model.schedule.pojos.Taskinfo;
  8. import com.heima.model.schedule.pojos.TaskinfoLogs;
  9. import com.heima.schedule.mapper.TaskinfoLogsMapper;
  10. import com.heima.schedule.mapper.TaskinfoMapper;
  11. import com.heima.schedule.service.TaskService;
  12. import lombok.extern.slf4j.Slf4j;
  13. import org.springframework.beans.BeanUtils;
  14. import org.springframework.beans.factory.annotation.Autowired;
  15. import org.springframework.stereotype.Service;
  16. import org.springframework.transaction.annotation.Transactional;
  17.  
  18. import java.util.Calendar;
  19. import java.util.Date;
  20.  
  21. @Service
  22. @Transactional
  23. @Slf4j
  24. public class TaskServiceImpl implements TaskService {
  25.     /**
  26.      * 添加延迟任务
  27.      *
  28.      * @param task
  29.      * @return
  30.      */
  31.     @Override
  32.     public long addTask(Task task) {
  33.         //1.添加任务到数据库中
  34.  
  35.         boolean success = addTaskToDb(task);
  36.  
  37.         if (success) {
  38.             //2.添加任务到redis
  39.             addTaskToCache(task);
  40.         }
  41.  
  42.  
  43.         return task.getTaskId();
  44.     }
  45.  
  46.     @Autowired
  47.     private CacheService cacheService;
  48.  
  49.     /**
  50.      * 把任务添加到redis中
  51.      *
  52.      * @param task
  53.      */
  54.     private void addTaskToCache(Task task) {
  55.  
  56.         String key = task.getTaskType() + "_" + task.getPriority();
  57.  
  58.         //获取5分钟之后的时间  毫秒值
  59.         Calendar calendar = Calendar.getInstance();
  60.         calendar.add(Calendar.MINUTE, 5);
  61.         long nextScheduleTime = calendar.getTimeInMillis();
  62.  
  63.         //2.1 如果任务的执行时间小于等于当前时间，存入list
  64.         if (task.getExecuteTime() <= System.currentTimeMillis()) {
  65.             cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
  66.         } else if (task.getExecuteTime() <= nextScheduleTime) {
  67.             //2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中
  68.             cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
  69.         }
  70.  
  71.  
  72.     }
  73.  
  74.     @Autowired
  75.     private TaskinfoMapper taskinfoMapper;
  76.  
  77.     @Autowired
  78.     private TaskinfoLogsMapper taskinfoLogsMapper;
  79.  
  80.     /**
  81.      * 添加任务到数据库中
  82.      *
  83.      * @param task
  84.      * @return
  85.      */
  86.     private boolean addTaskToDb(Task task) {
  87.  
  88.         boolean flag = false;
  89.  
  90.         try {
  91.             //保存任务表
  92.             Taskinfo taskinfo = new Taskinfo();
  93.             BeanUtils.copyProperties(task, taskinfo);
  94.             taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
  95.             taskinfoMapper.insert(taskinfo);
  96.  
  97.             //设置taskID
  98.             task.setTaskId(taskinfo.getTaskId());
  99.  
  100.             //保存任务日志数据
  101.             TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
  102.             BeanUtils.copyProperties(taskinfo, taskinfoLogs);
  103.             taskinfoLogs.setVersion(1);
  104.             taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
  105.             taskinfoLogsMapper.insert(taskinfoLogs);
  106.  
  107.             flag = true;
  108.         } catch (Exception e) {
  109.             e.printStackTrace();
  110.         }
  111.  
  112.         return flag;
  113.     }
  114. }

ScheduleConstants常量类

  1. package com.heima.common.constants;
  2.  
  3. public class ScheduleConstants {
  4.  
  5.     //task状态
  6.     public static final int SCHEDULED=0;   //初始化状态
  7.  
  8.     public static final int EXECUTED=1;       //已执行状态
  9.  
  10.     public static final int CANCELLED=2;   //已取消状态
  11.  
  12.     public static String FUTURE="future_";   //未来数据key前缀
  13.  
  14.     public static String TOPIC="topic_";     //当前数据key前缀
  15. }

5.取消任务

例如:第三接口网络不通,使用延迟任务进行重试,当达到阈值以后,取消任务.

在TaskService中添加方法

  1. /**
  2.      * 取消任务
  3.      * @param taskId        任务id
  4.      * @return              取消结果
  5.      */
  6. public boolean cancelTask(long taskId);

 实现

  1. /**
  2.      * 取消任务
  3.      * @param taskId
  4.      * @return
  5.      */
  6. @Override
  7. public boolean cancelTask(long taskId) {
  8.  
  9.     boolean flag = false;
  10.  
  11.     //删除任务，更新日志
  12.     Task task = updateDb(taskId,ScheduleConstants.EXECUTED);
  13.  
  14.     //删除redis的数据
  15.     if(task != null){
  16.         removeTaskFromCache(task);
  17.         flag = true;
  18.     }
  19.  
  20.  
  21.  
  22.     return false;
  23. }
  24.  
  25. /**
  26.      * 删除redis中的任务数据
  27.      * @param task
  28.      */
  29. private void removeTaskFromCache(Task task) {
  30.  
  31.     String key = task.getTaskType()+"_"+task.getPriority();
  32.  
  33.     if(task.getExecuteTime()<=System.currentTimeMillis()){
  34.         cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(task));
  35.     }else {
  36.         cacheService.zRemove(ScheduleConstants.FUTURE+key, JSON.toJSONString(task));
  37.     }
  38. }
  39.  
  40. /**
  41.      * 删除任务，更新任务日志状态
  42.      * @param taskId
  43.      * @param status
  44.      * @return
  45.      */
  46. private Task updateDb(long taskId, int status) {
  47.     Task task = null;
  48.     try {
  49.         //删除任务
  50.         taskinfoMapper.deleteById(taskId);
  51.  
  52.         TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
  53.         taskinfoLogs.setStatus(status);
  54.         taskinfoLogsMapper.updateById(taskinfoLogs);
  55.  
  56.         task = new Task();
  57.         BeanUtils.copyProperties(taskinfoLogs,task);
  58.         task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
  59.     }catch (Exception e){
  60.         log.error("task cancel exception taskid={}",taskId);
  61.     }
  62.  
  63.     return task;
  64.  
  65. }

6.消费任务

实现思路

在TaskService中添加方法

  1. /**
  2.  * 按照类型和优先级来拉取任务
  3.  * @param type
  4.  * @param priority
  5.  * @return
  6.  */
  7. public Task poll(int type,int priority);

 实现

  1. /**
  2.      * 按照类型和优先级拉取任务
  3.      * @return
  4.      */
  5. @Override
  6. public Task poll(int type,int priority) {
  7.     Task task = null;
  8.     try {
  9.         String key = type+"_"+priority;
  10.         String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
  11.         if(StringUtils.isNotBlank(task_json)){
  12.             task = JSON.parseObject(task_json, Task.class);
  13.             //更新数据库信息
  14.             updateDb(task.getTaskId(),ScheduleConstants.EXECUTED);
  15.         }
  16.     }catch (Exception e){
  17.         e.printStackTrace();
  18.         log.error("poll task exception");
  19.     }
  20.  
  21.     return task;
  22. }

7.定时任务刷新未来数据

 获取zset中所有key的两种方案

代码案例：

  1. @Test
  2. public void testKeys(){
  3.     //第一种方式,不建议使用
  4.     Set<String> keys = cacheService.keys("future_*");
  5.     System.out.println(keys);
  6.     //第二种方式,一次获取十个
  7.     Set<String> scan = cacheService.scan("future_*");
  8.     System.out.println(scan);
  9. }

返回结果

8.如何同步数据?

未来数据刷新-redis管道 性能很差,因为客户端要不断的创建连接

 redis管道技术

 测试案例对比：

  1. //耗时6151
  2. @Test
  3. public  void testPiple1(){
  4.     long start =System.currentTimeMillis();
  5.     for (int i = 0; i <10000 ; i++) {
  6.         Task task = new Task();
  7.         task.setTaskType(1001);
  8.         task.setPriority(1);
  9.         task.setExecuteTime(new Date().getTime());
  10.         cacheService.lLeftPush("1001_1", JSON.toJSONString(task));
  11.     }
  12.     System.out.println("耗时"+(System.currentTimeMillis()- start));
  13. }
  14.  
  15.  
  16. @Test
  17. public void testPiple2(){
  18.     long start  = System.currentTimeMillis();
  19.     //使用管道技术
  20.     List<Object> objectList = cacheService.getstringRedisTemplate().executePipelined(new RedisCallback<Object>() {
  21.         @Nullable
  22.         @Override
  23.         public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
  24.             for (int i = 0; i <10000 ; i++) {
  25.                 Task task = new Task();
  26.                 task.setTaskType(1001);
  27.                 task.setPriority(1);
  28.                 task.setExecuteTime(new Date().getTime());
  29.                 redisConnection.lPush("1001_1".getBytes(), JSON.toJSONString(task).getBytes());
  30.             }
  31.             return null;
  32.         }
  33.     });
  34.     System.out.println("使用管道技术执行10000次自增操作共耗时:"+(System.currentTimeMillis()-start)+"毫秒");
  35. }

9.使用定时任务刷新未来数据

此处用的单体定时任务,具体可根据项目使用分布式定时任务同步

在TaskService中添加方法

  1. @Scheduled(cron = "0 */1 * * * ?")
  2. public void refresh() {
  3.     System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
  4.  
  5.     // 获取所有未来数据集合的key值
  6.     Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*
  7.     for (String futureKey : futureKeys) { // future_250_250
  8.  
  9.         String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
  10.         //获取该组key下当前需要消费的任务数据
  11.         Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
  12.         if (!tasks.isEmpty()) {
  13.             //将这些任务数据添加到消费者队列中
  14.             cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
  15.             System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
  16.         }
  17.     }
  18. }

 10.分布式锁解决集群下单体定时任务方法抢占执行

启动两台heima-leadnews-schedule服务，每台服务都会去执行refresh定时任务方法

分布式锁：控制分布式系统有序的去对共享资源进行操作，通过互斥来保证数据的一致性。

解决方案：  

redis分布式锁

sexnx （SET if Not eXists） 命令在指定的 key 不存在时，为 key 设置指定的值。

这种加锁的思路是，如果 key 不存在则为 key 设置 value，如果 key 已存在则 SETNX 命令不做任何操作

  • 客户端A请求服务器设置key的值，如果设置成功就表示加锁成功

  • 客户端B也去请求服务器设置key的值，如果返回失败，那么就代表加锁失败

  • 客户端A执行代码完成，删除锁

  • 客户端B在等待一段时间后再去请求设置key的值，设置成功

  • 客户端B执行代码完成，删除锁

此加锁方法在cacheService工具类中

修改未来数据定时刷新的方法，如下：

  1. /**
  2.  * 未来数据定时刷新
  3.  */
  4. @Scheduled(cron = "0 */1 * * * ?")
  5. public void refresh(){
  6.  
  7.     String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
  8.     if(StringUtils.isNotBlank(token)){
  9.         log.info("未来数据定时刷新---定时任务");
  10.  
  11.         //获取所有未来数据的集合key
  12.         Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
  13.         for (String futureKey : futureKeys) {//future_100_50
  14.  
  15.             //获取当前数据的key  topic
  16.             String topicKey = ScheduleConstants.TOPIC+futureKey.split(ScheduleConstants.FUTURE)[1];
  17.  
  18.             //按照key和分值查询符合条件的数据
  19.             Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
  20.  
  21.             //同步数据
  22.             if(!tasks.isEmpty()){
  23.                 cacheService.refreshWithPipeline(futureKey,topicKey,tasks);
  24.                 log.info("成功的将"+futureKey+"刷新到了"+topicKey);
  25.             }
  26.         }
  27.     }
  28. }

10.数据库任务定时同步到redis

 第一步:

  1. @Scheduled(cron = "0 */5 * * * ?")
  2. @PostConstruct //只要微服务启动就会做一个同步
  3. public void reloadData() {
  4.     clearCache();
  5.     log.info("数据库数据同步到缓存");
  6.     Calendar calendar = Calendar.getInstance();
  7.     calendar.add(Calendar.MINUTE, 5);
  8.  
  9.     //查看小于未来5分钟的所有任务
  10.     List<Taskinfo> allTasks = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime,calendar.getTime()));
  11.     if(allTasks != null && allTasks.size() > 0){
  12.         for (Taskinfo taskinfo : allTasks) {
  13.             Task task = new Task();
  14.             BeanUtils.copyProperties(taskinfo,task);
  15.             task.setExecuteTime(taskinfo.getExecuteTime().getTime());
  16.             addTaskToCache(task);
  17.         }
  18.     }
  19. }
  20.  
  21. private void clearCache(){
  22.     // 删除缓存中未来数据集合和当前消费者队列的所有key
  23.     Set<String> futurekeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_
  24.     Set<String> topickeys = cacheService.scan(ScheduleConstants.TOPIC + "*");// topic_
  25.     cacheService.delete(futurekeys);
  26.     cacheService.delete(topickeys);
  27. }

11.延迟任务对外提供接口,提供远程fegin接口

提供远程的feign接口，在heima-leadnews-feign-api编写类如下：

  1. package com.heima.apis.schedule;
  2.  
  3. import com.heima.model.common.dtos.ResponseResult;
  4. import com.heima.model.schedule.dtos.Task;
  5. import org.springframework.cloud.openfeign.FeignClient;
  6. import org.springframework.web.bind.annotation.GetMapping;
  7. import org.springframework.web.bind.annotation.PathVariable;
  8. import org.springframework.web.bind.annotation.PostMapping;
  9. import org.springframework.web.bind.annotation.RequestBody;
  10.  
  11. @FeignClient("leadnews-schedule")
  12. public interface IScheduleClient {
  13.  
  14.     /**
  15.      * 添加任务
  16.      * @param task   任务对象
  17.      * @return       任务id
  18.      */
  19.     @PostMapping("/api/v1/task/add")
  20.     public ResponseResult  addTask(@RequestBody Task task);
  21.  
  22.     /**
  23.      * 取消任务
  24.      * @param taskId        任务id
  25.      * @return              取消结果
  26.      */
  27.     @GetMapping("/api/v1/task/cancel/{taskId}")
  28.     public ResponseResult cancelTask(@PathVariable("taskId") long taskId);
  29.  
  30.     /**
  31.      * 按照类型和优先级来拉取任务
  32.      * @param type
  33.      * @param priority
  34.      * @return
  35.      */
  36.     @GetMapping("/api/v1/task/poll/{type}/{priority}")
  37.     public ResponseResult poll(@PathVariable("type") int type,@PathVariable("priority")  int priority);
  38. }

在heima-leadnews-schedule微服务下提供对应的实现

  1. package com.heima.schedule.feign;
  2.  
  3. import com.heima.apis.schedule.IScheduleClient;
  4. import com.heima.model.common.dtos.ResponseResult;
  5. import com.heima.model.schedule.dtos.Task;
  6. import com.heima.schedule.service.TaskService;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.web.bind.annotation.*;
  9.  
  10.  
  11. @RestController
  12. public class ScheduleClient  implements IScheduleClient {
  13.  
  14.     @Autowired
  15.     private TaskService taskService;
  16.  
  17.     /**
  18.      * 添加任务
  19.      * @param task 任务对象
  20.      * @return 任务id
  21.      */
  22.     @PostMapping("/api/v1/task/add")
  23.     @Override
  24.     public ResponseResult addTask(@RequestBody Task task) {
  25.         return ResponseResult.okResult(taskService.addTask(task));
  26.     }
  27.  
  28.     /**
  29.      * 取消任务
  30.      * @param taskId 任务id
  31.      * @return 取消结果
  32.      */
  33.     @GetMapping("/api/v1/task/cancel/{taskId}")
  34.     @Override
  35.     public ResponseResult cancelTask(@PathVariable("taskId") long taskId) {
  36.         return ResponseResult.okResult(taskService.cancelTask(taskId));
  37.     }
  38.  
  39.     /**
  40.      * 按照类型和优先级来拉取任务
  41.      * @param type
  42.      * @param priority
  43.      * @return
  44.      */
  45.     @GetMapping("/api/v1/task/poll/{type}/{priority}")
  46.     @Override
  47.     public ResponseResult poll(@PathVariable("type") int type, @PathVariable("priority") int priority) {
  48.         return ResponseResult.okResult(taskService.poll(type,priority));
  49.     }
  50. }

12.发布文章集成添加延迟队列接口

在创建WmNewsTaskService

  1. package com.heima.wemedia.service;
  2.  
  3. import com.heima.model.wemedia.pojos.WmNews;
  4.  
  5.  
  6. public interface WmNewsTaskService {
  7.  
  8.     /**
  9.      * 添加任务到延迟队列中
  10.      * @param id  文章的id
  11.      * @param publishTime  发布的时间  可以做为任务的执行时间
  12.      */
  13.     public void addNewsToTask(Integer id, Date publishTime);
  14.  
  15.  
  16. }

实现：

  1. package com.heima.wemedia.service.impl;
  2.  
  3. import com.heima.apis.schedule.IScheduleClient;
  4. import com.heima.model.common.enums.TaskTypeEnum;
  5. import com.heima.model.schedule.dtos.Task;
  6. import com.heima.model.wemedia.pojos.WmNews;
  7. import com.heima.utils.common.ProtostuffUtil;
  8. import com.heima.wemedia.service.WmNewsTaskService;
  9. import lombok.SneakyThrows;
  10. import lombok.extern.slf4j.Slf4j;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.scheduling.annotation.Async;
  13. import org.springframework.stereotype.Service;
  14.  
  15.  
  16. @Service
  17. @Slf4j
  18. public class WmNewsTaskServiceImpl  implements WmNewsTaskService {
  19.  
  20.  
  21.     @Autowired
  22.     private IScheduleClient scheduleClient;
  23.  
  24.     /**
  25.      * 添加任务到延迟队列中
  26.      * @param id          文章的id
  27.      * @param publishTime 发布的时间  可以做为任务的执行时间
  28.      */
  29.     @Override
  30.     @Async
  31.     public void addNewsToTask(Integer id, Date publishTime) {
  32.  
  33.         log.info("添加任务到延迟服务中----begin");
  34.  
  35.         Task task = new Task();
  36.         task.setExecuteTime(publishTime.getTime());
  37.         task.setTaskType(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType());
  38.         task.setPriority(TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
  39.         WmNews wmNews = new WmNews();
  40.         wmNews.setId(id);
  41.         task.setParameters(ProtostuffUtil.serialize(wmNews));
  42.  
  43.         scheduleClient.addTask(task);
  44.  
  45.         log.info("添加任务到延迟服务中----end");
  46.  
  47.     }
  48.     
  49. }

枚举类：

  1. package com.heima.model.common.enums;
  2.  
  3. import lombok.AllArgsConstructor;
  4. import lombok.Getter;
  5.  
  6. @Getter
  7. @AllArgsConstructor
  8. public enum TaskTypeEnum {
  9.  
  10.     NEWS_SCAN_TIME(1001, 1,"文章定时审核"),
  11.     REMOTEERROR(1002, 2,"第三方接口调用失败，重试");
  12.     private final int taskType; //对应具体业务
  13.     private final int priority; //业务不同级别
  14.     private final String desc; //描述信息
  15. }

序列化工具对比

  • JdkSerialize：java内置的序列化能将实现了Serilazable接口的对象进行序列化和反序列化， ObjectOutputStream的writeObject()方法可序列化对象生成字节数组

  • Protostuff：google开源的protostuff采用更为紧凑的二进制数组，表现更加优异，然后使用protostuff的编译工具生成pojo类

拷贝资料中的两个类到heima-leadnews-utils下

Protostuff需要引导依赖：

  1. <dependency>
  2.     <groupId>io.protostuff</groupId>
  3.     <artifactId>protostuff-core</artifactId>
  4.     <version>1.6.0</version>
  5. </dependency>
  6.  
  7. <dependency>
  8.     <groupId>io.protostuff</groupId>
  9.     <artifactId>protostuff-runtime</artifactId>
  10.     <version>1.6.0</version>
  11. </dependency>

修改发布文章代码：

把之前的异步调用修改为调用延迟任务

  1. @Autowired
  2. private WmNewsTaskService wmNewsTaskService;
  3.  
  4. /**
  5.      * 发布修改文章或保存为草稿
  6.      * @param dto
  7.      * @return
  8.      */
  9. @Override
  10. public ResponseResult submitNews(WmNewsDto dto) {
  11.  
  12.     //0.条件判断
  13.     if(dto == null || dto.getContent() == null){
  14.         return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
  15.     }
  16.  
  17.     //1.保存或修改文章
  18.  
  19.     WmNews wmNews = new WmNews();
  20.     //属性拷贝 属性名词和类型相同才能拷贝
  21.     BeanUtils.copyProperties(dto,wmNews);
  22.     //封面图片  list---> string
  23.     if(dto.getImages() != null && dto.getImages().size() > 0){
  24.         //[1dddfsd.jpg,sdlfjldk.jpg]-->   1dddfsd.jpg,sdlfjldk.jpg
  25.         String imageStr = StringUtils.join(dto.getImages(), ",");
  26.         wmNews.setImages(imageStr);
  27.     }
  28.     //如果当前封面类型为自动 -1
  29.     if(dto.getType().equals(WemediaConstants.WM_NEWS_TYPE_AUTO)){
  30.         wmNews.setType(null);
  31.     }
  32.  
  33.     saveOrUpdateWmNews(wmNews);
  34.  
  35.     //2.判断是否为草稿  如果为草稿结束当前方法
  36.     if(dto.getStatus().equals(WmNews.Status.NORMAL.getCode())){
  37.         return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
  38.     }
  39.  
  40.     //3.不是草稿，保存文章内容图片与素材的关系
  41.     //获取到文章内容中的图片信息
  42.     List<String> materials =  ectractUrlInfo(dto.getContent());
  43.     saveRelativeInfoForContent(materials,wmNews.getId());
  44.  
  45.     //4.不是草稿，保存文章封面图片与素材的关系，如果当前布局是自动，需要匹配封面图片
  46.     saveRelativeInfoForCover(dto,wmNews,materials);
  47.  
  48.     //审核文章
  49.     //        wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
  50.     wmNewsTaskService.addNewsToTask(wmNews.getId(),wmNews.getPublishTime());
  51.  
  52.     return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
  53.  
  54. }

消费任务进行审核文章

WmNewsTaskService中添加方法

  1. /**
  2.  * 消费延迟队列数据
  3.  */
  4. public void scanNewsByTask();

实现

  1. @Autowired
  2. private WmNewsAutoScanServiceImpl wmNewsAutoScanService;
  3.  
  4. /**
  5.      * 消费延迟队列数据
  6.      */
  7. @Scheduled(fixedRate = 1000)
  8. @Override
  9. @SneakyThrows
  10. public void scanNewsByTask() {
  11.  
  12.     log.info("文章审核---消费任务执行---begin---");
  13.  
  14.     ResponseResult responseResult = scheduleClient.poll(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType(), TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
  15.     if(responseResult.getCode().equals(200) && responseResult.getData() != null){
  16.         String json_str = JSON.toJSONString(responseResult.getData());
  17.         Task task = JSON.parseObject(json_str, Task.class);
  18.         byte[] parameters = task.getParameters();
  19.         WmNews wmNews = ProtostuffUtil.deserialize(parameters, WmNews.class);
  20.         System.out.println(wmNews.getId()+"-----------");
  21.         wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
  22.     }
  23.     log.info("文章审核---消费任务执行---end---");
  24. }

在WemediaApplication自媒体的引导类中添加开启任务调度注解@EnableScheduling

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/125696?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号