redis实现消息延迟队列

业务场景

在很多软件系统功能中都会出现定时任务的业务场景,比如提前点单,比如定时发布动态,文章等而出现这样的的定时的任务为延迟队任务

代码模块

任务的持久化一般都需要建立一个任务表和任务日志表,避免宕机导致任务失效,先新建立一个数据库,创建基本的任务表和任务日志表,其中的parameter设置为longbite,是防止消息使用json传递消耗较大,节省资源
在这里插入图片描述

参数是在任务中存储操作任务需要的关键信息,设置为比特位是更节省资源

模拟业务文章定时发布,前端传递了一个文章对象，其中包含了预期发布的时间在这里插入图片描述
传递的文章dto中包含了对应的文章信息,我们需要做的就是根据审核内容进行保存在文章表,设置是否上架到用户端,如果携带的发布时间比现在要大,说明是延迟任务,那么此时就涉及到延迟任务的实现

代码片段

controller

主要是接收应用的http请求

@RestController
@RequestMapping("/api/v1/news")
public class NewsController {@AutowiredNewsService wmNewsService;@PostMapping("/list")public ResponseResult findAll(@RequestBody WmNewsPageReqDto dto){return  wmNewsService.findAll(dto);}
/**
* 文章的提交
*/@PostMapping("/submit")public ResponseResult submit(@RequestBody WmNewsDto dto){return wmNewsService.submitNews(dto);}}

业务层Service

用于处理文章提交的逻辑


@Service
@Slf4j
@Transactional
public class WmNewsServiceImpl extends ServiceImpl<WmNewsMapper, WmNews> implements WmNewsService {/**
* 自动扫描审核文章业务 这里不是重点不用关注
/@AutowiredWmTaskService taskService;///*** 发布修改文章或保存为草稿* @param dto* @return*/@Overridepublic ResponseResult submitNews(WmNewsDto dto) {//0.条件判断if(dto == null || dto.getContent() == null){return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);}//1.保存或修改文章WmNews wmNews = new WmNews();//属性拷贝 属性名词和类型相同才能拷贝BeanUtils.copyProperties(dto,wmNews);//封面图片  list---> stringif(dto.getImages() != null && dto.getImages().size() > 0){//[1dddfsd.jpg,sdlfjldk.jpg]-->   1dddfsd.jpg,sdlfjldk.jpgString imageStr = StringUtils.join(dto.getImages(), ",");wmNews.setImages(imageStr);}//如果当前封面类型为自动 -1if(dto.getType().equals(WemediaConstants.WM_NEWS_TYPE_AUTO)){wmNews.setType(null);}saveOrUpdateWmNews(wmNews);//2.判断是否为草稿  如果为草稿结束当前方法if(dto.getStatus().equals(WmNews.Status.NORMAL.getCode())){return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);}//3.不是草稿，保存文章内容图片与素材的关系//获取到文章内容中的图片信息List<String> materials =  ectractUrlInfo(dto.getContent());saveRelativeInfoForContent(materials,wmNews.getId());//4.不是草稿，保存文章封面图片与素材的关系，如果当前布局是自动，需要匹配封面图片saveRelativeInfoForCover(dto,wmNews,materials);/*
*
*上面都不用看,是对文章的处理逻辑,
*当对文章完成处理后
*///上面的一大堆都是对文章的处理逻辑if (wmNews.getId()!=null){taskService.addNewsToTask(wmNews.getId(),wmNews.getPublishTime());}return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);}}}

文章处理的逻辑并不重要,重要的是完成对延迟任务的实现

所以重点关注定时任务,原本的模块是管理端完成文章的处理–>客户端的文章上架(保存到数据库还是修改库不需要关注),现在变成了:管理端完成了–>定时任务的处理–>时间到了–>客户端文章的处理,所以我们需要关注的就是任务处理的部分,因为这里假设的场景是微服务的场景下,所以就需要新建一个定时任务模块,并且将上述的taskService改为feign模块的远程调用接口

建立一个schedule 模块

添加redis的依赖

     <!--spring data redis & cache--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- redis依赖commons-pool 这个依赖一定要添加 --><dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-cache</artifactId></dependency>

2.实现思路在这里插入图片描述
添加一个任务对象到数据库,然后进行判断,如果执行时间<=当前的时间说明需要放入list队列,等待被消费者消费,如果大于当前时间,但是小于设置预设时间(如果大于预设时间,说明任务举例处理还很遥远, 不对其进行处理避免多余的io操作),说明是将要消费任务,使用set数据类型进行存储,并且使用时间错作为score,这样做到排序的效果

建立一个缓存service 集成spring cache

@Component
public class CacheService extends CachingConfigurerSupport {@Autowiredprivate StringRedisTemplate stringRedisTemplate;public StringRedisTemplate getstringRedisTemplate() {return this.stringRedisTemplate;}/** -------------------key相关操作--------------------- *//*** 删除key** @param key*/public void delete(String key) {stringRedisTemplate.delete(key);}/*** 批量删除key** @param keys*/public void delete(Collection<String> keys) {stringRedisTemplate.delete(keys);}/*** 序列化key** @param key* @return*/public byte[] dump(String key) {return stringRedisTemplate.dump(key);}/*** 是否存在key** @param key* @return*/public Boolean exists(String key) {return stringRedisTemplate.hasKey(key);}/*** 设置过期时间** @param key* @param timeout* @param unit* @return*/public Boolean expire(String key, long timeout, TimeUnit unit) {return stringRedisTemplate.expire(key, timeout, unit);}/*** 设置过期时间** @param key* @param date* @return*/public Boolean expireAt(String key, Date date) {return stringRedisTemplate.expireAt(key, date);}/*** 查找匹配的key** @param pattern* @return*/public Set<String> keys(String pattern) {return stringRedisTemplate.keys(pattern);}/*** 将当前数据库的 key 移动到给定的数据库 db 当中** @param key* @param dbIndex* @return*/public Boolean move(String key, int dbIndex) {return stringRedisTemplate.move(key, dbIndex);}/*** 移除 key 的过期时间，key 将持久保持** @param key* @return*/public Boolean persist(String key) {return stringRedisTemplate.persist(key);}/*** 返回 key 的剩余的过期时间** @param key* @param unit* @return*/public Long getExpire(String key, TimeUnit unit) {return stringRedisTemplate.getExpire(key, unit);}/*** 返回 key 的剩余的过期时间** @param key* @return*/public Long getExpire(String key) {return stringRedisTemplate.getExpire(key);}/*** 从当前数据库中随机返回一个 key** @return*/public String randomKey() {return stringRedisTemplate.randomKey();}/*** 修改 key 的名称** @param oldKey* @param newKey*/public void rename(String oldKey, String newKey) {stringRedisTemplate.rename(oldKey, newKey);}/*** 仅当 newkey 不存在时，将 oldKey 改名为 newkey** @param oldKey* @param newKey* @return*/public Boolean renameIfAbsent(String oldKey, String newKey) {return stringRedisTemplate.renameIfAbsent(oldKey, newKey);}/*** 返回 key 所储存的值的类型** @param key* @return*/public DataType type(String key) {return stringRedisTemplate.type(key);}/** -------------------string相关操作--------------------- *//*** 设置指定 key 的值* @param key* @param value*/public void set(String key, String value) {stringRedisTemplate.opsForValue().set(key, value);}/*** 获取指定 key 的值* @param key* @return*/public String get(String key) {return stringRedisTemplate.opsForValue().get(key);}/*** 返回 key 中字符串值的子字符* @param key* @param start* @param end* @return*/public String getRange(String key, long start, long end) {return stringRedisTemplate.opsForValue().get(key, start, end);}/*** 将给定 key 的值设为 value ，并返回 key 的旧值(old value)** @param key* @param value* @return*/public String getAndSet(String key, String value) {return stringRedisTemplate.opsForValue().getAndSet(key, value);}/*** 对 key 所储存的字符串值，获取指定偏移量上的位(bit)** @param key* @param offset* @return*/public Boolean getBit(String key, long offset) {return stringRedisTemplate.opsForValue().getBit(key, offset);}/*** 批量获取** @param keys* @return*/public List<String> multiGet(Collection<String> keys) {return stringRedisTemplate.opsForValue().multiGet(keys);}/*** 设置ASCII码, 字符串'a'的ASCII码是97, 转为二进制是'01100001', 此方法是将二进制第offset位值变为value** @param key* @param* @param value*            值,true为1, false为0* @return*/public boolean setBit(String key, long offset, boolean value) {return stringRedisTemplate.opsForValue().setBit(key, offset, value);}/*** 将值 value 关联到 key ，并将 key 的过期时间设为 timeout** @param key* @param value* @param timeout*            过期时间* @param unit*            时间单位, 天:TimeUnit.DAYS 小时:TimeUnit.HOURS 分钟:TimeUnit.MINUTES*            秒:TimeUnit.SECONDS 毫秒:TimeUnit.MILLISECONDS*/public void setEx(String key, String value, long timeout, TimeUnit unit) {stringRedisTemplate.opsForValue().set(key, value, timeout, unit);}/*** 只有在 key 不存在时设置 key 的值** @param key* @param value* @return 之前已经存在返回false,不存在返回true*/public boolean setIfAbsent(String key, String value) {return stringRedisTemplate.opsForValue().setIfAbsent(key, value);}/*** 用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始** @param key* @param value* @param offset*            从指定位置开始覆写*/public void setRange(String key, String value, long offset) {stringRedisTemplate.opsForValue().set(key, value, offset);}/*** 获取字符串的长度** @param key* @return*/public Long size(String key) {return stringRedisTemplate.opsForValue().size(key);}/*** 批量添加** @param maps*/public void multiSet(Map<String, String> maps) {stringRedisTemplate.opsForValue().multiSet(maps);}/*** 同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在** @param maps* @return 之前已经存在返回false,不存在返回true*/public boolean multiSetIfAbsent(Map<String, String> maps) {return stringRedisTemplate.opsForValue().multiSetIfAbsent(maps);}/*** 增加(自增长), 负数则为自减** @param key* @param* @return*/public Long incrBy(String key, long increment) {return stringRedisTemplate.opsForValue().increment(key, increment);}/**** @param key* @param* @return*/public Double incrByFloat(String key, double increment) {return stringRedisTemplate.opsForValue().increment(key, increment);}/*** 追加到末尾** @param key* @param value* @return*/public Integer append(String key, String value) {return stringRedisTemplate.opsForValue().append(key, value);}/** -------------------hash相关操作------------------------- *//*** 获取存储在哈希表中指定字段的值** @param key* @param field* @return*/public Object hGet(String key, String field) {return stringRedisTemplate.opsForHash().get(key, field);}/*** 获取所有给定字段的值** @param key* @return*/public Map<Object, Object> hGetAll(String key) {return stringRedisTemplate.opsForHash().entries(key);}/*** 获取所有给定字段的值** @param key* @param fields* @return*/public List<Object> hMultiGet(String key, Collection<Object> fields) {return stringRedisTemplate.opsForHash().multiGet(key, fields);}public void hPut(String key, String hashKey, String value) {stringRedisTemplate.opsForHash().put(key, hashKey, value);}public void hPutAll(String key, Map<String, String> maps) {stringRedisTemplate.opsForHash().putAll(key, maps);}/*** 仅当hashKey不存在时才设置** @param key* @param hashKey* @param value* @return*/public Boolean hPutIfAbsent(String key, String hashKey, String value) {return stringRedisTemplate.opsForHash().putIfAbsent(key, hashKey, value);}/*** 加锁** @param name* @param expire* @return*/public String tryLock(String name, long expire) {name = name + "_lock";String token = UUID.randomUUID().toString();RedisConnectionFactory factory = stringRedisTemplate.getConnectionFactory();RedisConnection conn = factory.getConnection();try {//参考redis命令：在redis中存入数据 数据明 uuid token 为锁的名字
//            因为这个数据的名字都是name+_locak 所以每次调用加锁方法setnx 只有一个成功调用的才能才能成功的保存的一个锁数据,其他的只能无法设置//set key value [EX seconds] [PX milliseconds] [NX|XX]Boolean result = conn.set(name.getBytes(),token.getBytes(),Expiration.from(expire, TimeUnit.MILLISECONDS),RedisStringCommands.SetOption.SET_IF_ABSENT //NX);if (result != null && result)return token;} finally {RedisConnectionUtils.releaseConnection(conn, factory,false);}return null;}/*** 删除一个或多个哈希表字段** @param key* @param fields* @return*/public Long hDelete(String key, Object... fields) {return stringRedisTemplate.opsForHash().delete(key, fields);}/*** 查看哈希表 key 中，指定的字段是否存在** @param key* @param field* @return*/public boolean hExists(String key, String field) {return stringRedisTemplate.opsForHash().hasKey(key, field);}/*** 为哈希表 key 中的指定字段的整数值加上增量 increment** @param key* @param field* @param increment* @return*/public Long hIncrBy(String key, Object field, long increment) {return stringRedisTemplate.opsForHash().increment(key, field, increment);}/*** 为哈希表 key 中的指定字段的整数值加上增量 increment** @param key* @param field* @param delta* @return*/public Double hIncrByFloat(String key, Object field, double delta) {return stringRedisTemplate.opsForHash().increment(key, field, delta);}/*** 获取所有哈希表中的字段** @param key* @return*/public Set<Object> hKeys(String key) {return stringRedisTemplate.opsForHash().keys(key);}/*** 获取哈希表中字段的数量** @param key* @return*/public Long hSize(String key) {return stringRedisTemplate.opsForHash().size(key);}/*** 获取哈希表中所有值** @param key* @return*/public List<Object> hValues(String key) {return stringRedisTemplate.opsForHash().values(key);}/*** 迭代哈希表中的键值对** @param key* @param options* @return*/public Cursor<Map.Entry<Object, Object>> hScan(String key, ScanOptions options) {return stringRedisTemplate.opsForHash().scan(key, options);}/** ------------------------list相关操作---------------------------- *//*** 通过索引获取列表中的元素** @param key* @param index* @return*/public String lIndex(String key, long index) {return stringRedisTemplate.opsForList().index(key, index);}/*** 获取列表指定范围内的元素** @param key* @param start*            开始位置, 0是开始位置* @param end*            结束位置, -1返回所有* @return*/public List<String> lRange(String key, long start, long end) {return stringRedisTemplate.opsForList().range(key, start, end);}/*** 存储在list头部** @param key* @param value* @return*/public Long lLeftPush(String key, String value) {return stringRedisTemplate.opsForList().leftPush(key, value);}/**** @param key* @param value* @return*/public Long lLeftPushAll(String key, String... value) {return stringRedisTemplate.opsForList().leftPushAll(key, value);}/**** @param key* @param value* @return*/public Long lLeftPushAll(String key, Collection<String> value) {return stringRedisTemplate.opsForList().leftPushAll(key, value);}/*** 当list存在的时候才加入** @param key* @param value* @return*/public Long lLeftPushIfPresent(String key, String value) {return stringRedisTemplate.opsForList().leftPushIfPresent(key, value);}/*** 如果pivot存在,再pivot前面添加** @param key* @param pivot* @param value* @return*/public Long lLeftPush(String key, String pivot, String value) {return stringRedisTemplate.opsForList().leftPush(key, pivot, value);}/**** @param key* @param value* @return*/public Long lRightPush(String key, String value) {return stringRedisTemplate.opsForList().rightPush(key, value);}/**** @param key* @param value* @return*/public Long lRightPushAll(String key, String... value) {return stringRedisTemplate.opsForList().rightPushAll(key, value);}/**** @param key* @param value* @return*/public Long lRightPushAll(String key, Collection<String> value) {return stringRedisTemplate.opsForList().rightPushAll(key, value);}/*** 为已存在的列表添加值** @param key* @param value* @return*/public Long lRightPushIfPresent(String key, String value) {return stringRedisTemplate.opsForList().rightPushIfPresent(key, value);}/*** 在pivot元素的右边添加值** @param key* @param pivot* @param value* @return*/public Long lRightPush(String key, String pivot, String value) {return stringRedisTemplate.opsForList().rightPush(key, pivot, value);}/*** 通过索引设置列表元素的值** @param key* @param index*            位置* @param value*/public void lSet(String key, long index, String value) {stringRedisTemplate.opsForList().set(key, index, value);}/*** 移出并获取列表的第一个元素** @param key* @return 删除的元素*/public String lLeftPop(String key) {return stringRedisTemplate.opsForList().leftPop(key);}/*** 移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止** @param key* @param timeout*            等待时间* @param unit*            时间单位* @return*/public String lBLeftPop(String key, long timeout, TimeUnit unit) {return stringRedisTemplate.opsForList().leftPop(key, timeout, unit);}/*** 移除并获取列表最后一个元素** @param key* @return 删除的元素*/public String lRightPop(String key) {return stringRedisTemplate.opsForList().rightPop(key);}/*** 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止** @param key* @param timeout*            等待时间* @param unit*            时间单位* @return*/public String lBRightPop(String key, long timeout, TimeUnit unit) {return stringRedisTemplate.opsForList().rightPop(key, timeout, unit);}/*** 移除列表的最后一个元素，并将该元素添加到另一个列表并返回** @param sourceKey* @param destinationKey* @return*/public String lRightPopAndLeftPush(String sourceKey, String destinationKey) {return stringRedisTemplate.opsForList().rightPopAndLeftPush(sourceKey,destinationKey);}/*** 从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止** @param sourceKey* @param destinationKey* @param timeout* @param unit* @return*/public String lBRightPopAndLeftPush(String sourceKey, String destinationKey,long timeout, TimeUnit unit) {return stringRedisTemplate.opsForList().rightPopAndLeftPush(sourceKey,destinationKey, timeout, unit);}/*** 删除集合中值等于value得元素** @param key* @param index*            index=0, 删除所有值等于value的元素; index>0, 从头部开始删除第一个值等于value的元素;*            index<0, 从尾部开始删除第一个值等于value的元素;* @param value* @return*/public Long lRemove(String key, long index, String value) {return stringRedisTemplate.opsForList().remove(key, index, value);}/*** 裁剪list** @param key* @param start* @param end*/public void lTrim(String key, long start, long end) {stringRedisTemplate.opsForList().trim(key, start, end);}/*** 获取列表长度** @param key* @return*/public Long lLen(String key) {return stringRedisTemplate.opsForList().size(key);}/** --------------------set相关操作-------------------------- *//*** set添加元素** @param key* @param values* @return*/public Long sAdd(String key, String... values) {return stringRedisTemplate.opsForSet().add(key, values);}/*** set移除元素** @param key* @param values* @return*/public Long sRemove(String key, Object... values) {return stringRedisTemplate.opsForSet().remove(key, values);}/*** 移除并返回集合的一个随机元素** @param key* @return*/public String sPop(String key) {return stringRedisTemplate.opsForSet().pop(key);}/*** 将元素value从一个集合移到另一个集合** @param key* @param value* @param destKey* @return*/public Boolean sMove(String key, String value, String destKey) {return stringRedisTemplate.opsForSet().move(key, value, destKey);}/*** 获取集合的大小** @param key* @return*/public Long sSize(String key) {return stringRedisTemplate.opsForSet().size(key);}/*** 判断集合是否包含value** @param key* @param value* @return*/public Boolean sIsMember(String key, Object value) {return stringRedisTemplate.opsForSet().isMember(key, value);}/*** 获取两个集合的交集** @param key* @param otherKey* @return*/public Set<String> sIntersect(String key, String otherKey) {return stringRedisTemplate.opsForSet().intersect(key, otherKey);}/*** 获取key集合与多个集合的交集** @param key* @param otherKeys* @return*/public Set<String> sIntersect(String key, Collection<String> otherKeys) {return stringRedisTemplate.opsForSet().intersect(key, otherKeys);}/*** key集合与otherKey集合的交集存储到destKey集合中** @param key* @param otherKey* @param destKey* @return*/public Long sIntersectAndStore(String key, String otherKey, String destKey) {return stringRedisTemplate.opsForSet().intersectAndStore(key, otherKey,destKey);}/*** key集合与多个集合的交集存储到destKey集合中** @param key* @param otherKeys* @param destKey* @return*/public Long sIntersectAndStore(String key, Collection<String> otherKeys,String destKey) {return stringRedisTemplate.opsForSet().intersectAndStore(key, otherKeys,destKey);}/*** 获取两个集合的并集** @param key* @param otherKeys* @return*/public Set<String> sUnion(String key, String otherKeys) {return stringRedisTemplate.opsForSet().union(key, otherKeys);}/*** 获取key集合与多个集合的并集** @param key* @param otherKeys* @return*/public Set<String> sUnion(String key, Collection<String> otherKeys) {return stringRedisTemplate.opsForSet().union(key, otherKeys);}/*** key集合与otherKey集合的并集存储到destKey中** @param key* @param otherKey* @param destKey* @return*/public Long sUnionAndStore(String key, String otherKey, String destKey) {return stringRedisTemplate.opsForSet().unionAndStore(key, otherKey, destKey);}/*** key集合与多个集合的并集存储到destKey中** @param key* @param otherKeys* @param destKey* @return*/public Long sUnionAndStore(String key, Collection<String> otherKeys,String destKey) {return stringRedisTemplate.opsForSet().unionAndStore(key, otherKeys, destKey);}/*** 获取两个集合的差集** @param key* @param otherKey* @return*/public Set<String> sDifference(String key, String otherKey) {return stringRedisTemplate.opsForSet().difference(key, otherKey);}/*** 获取key集合与多个集合的差集** @param key* @param otherKeys* @return*/public Set<String> sDifference(String key, Collection<String> otherKeys) {return stringRedisTemplate.opsForSet().difference(key, otherKeys);}/*** key集合与otherKey集合的差集存储到destKey中** @param key* @param otherKey* @param destKey* @return*/public Long sDifference(String key, String otherKey, String destKey) {return stringRedisTemplate.opsForSet().differenceAndStore(key, otherKey,destKey);}/*** key集合与多个集合的差集存储到destKey中** @param key* @param otherKeys* @param destKey* @return*/public Long sDifference(String key, Collection<String> otherKeys,String destKey) {return stringRedisTemplate.opsForSet().differenceAndStore(key, otherKeys,destKey);}/*** 获取集合所有元素** @param key* @param* @param* @return*/public Set<String> setMembers(String key) {return stringRedisTemplate.opsForSet().members(key);}/*** 随机获取集合中的一个元素** @param key* @return*/public String sRandomMember(String key) {return stringRedisTemplate.opsForSet().randomMember(key);}/*** 随机获取集合中count个元素** @param key* @param count* @return*/public List<String> sRandomMembers(String key, long count) {return stringRedisTemplate.opsForSet().randomMembers(key, count);}/*** 随机获取集合中count个元素并且去除重复的** @param key* @param count* @return*/public Set<String> sDistinctRandomMembers(String key, long count) {return stringRedisTemplate.opsForSet().distinctRandomMembers(key, count);}/**** @param key* @param options* @return*/public Cursor<String> sScan(String key, ScanOptions options) {return stringRedisTemplate.opsForSet().scan(key, options);}/**------------------zSet相关操作--------------------------------*//*** 添加元素,有序集合是按照元素的score值由小到大排列** @param key* @param value* @param score* @return*/public Boolean zAdd(String key, String value, double score) {return stringRedisTemplate.opsForZSet().add(key, value, score);}/**** @param key* @param values* @return*/public Long zAdd(String key, Set<TypedTuple<String>> values) {return stringRedisTemplate.opsForZSet().add(key, values);}/**** @param key* @param values* @return*/public Long zRemove(String key, Object... values) {return stringRedisTemplate.opsForZSet().remove(key, values);}public Long zRemove(String key, Collection<String> values) {if(values!=null&&!values.isEmpty()){Object[] objs = values.toArray(new Object[values.size()]);return stringRedisTemplate.opsForZSet().remove(key, objs);}return 0L;}/*** 增加元素的score值，并返回增加后的值** @param key* @param value* @param delta* @return*/public Double zIncrementScore(String key, String value, double delta) {return stringRedisTemplate.opsForZSet().incrementScore(key, value, delta);}/*** 返回元素在集合的排名,有序集合是按照元素的score值由小到大排列** @param key* @param value* @return 0表示第一位*/public Long zRank(String key, Object value) {return stringRedisTemplate.opsForZSet().rank(key, value);}/*** 返回元素在集合的排名,按元素的score值由大到小排列** @param key* @param value* @return*/public Long zReverseRank(String key, Object value) {return stringRedisTemplate.opsForZSet().reverseRank(key, value);}/*** 获取集合的元素, 从小到大排序** @param key* @param start*            开始位置* @param end*            结束位置, -1查询所有* @return*/public Set<String> zRange(String key, long start, long end) {return stringRedisTemplate.opsForZSet().range(key, start, end);}/*** 获取zset集合的所有元素, 从小到大排序**/public Set<String> zRangeAll(String key) {return zRange(key,0,-1);}/*** 获取集合元素, 并且把score值也获取** @param key* @param start* @param end* @return*/public Set<TypedTuple<String>> zRangeWithScores(String key, long start,long end) {return stringRedisTemplate.opsForZSet().rangeWithScores(key, start, end);}/*** 根据Score值查询集合元素** @param key* @param min*            最小值* @param max*            最大值* @return*/public Set<String> zRangeByScore(String key, double min, double max) {return stringRedisTemplate.opsForZSet().rangeByScore(key, min, max);}/*** 根据Score值查询集合元素, 从小到大排序** @param key* @param min*            最小值* @param max*            最大值* @return*/public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,double min, double max) {return stringRedisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max);}/**** @param key* @param min* @param max* @param start* @param end* @return*/public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,double min, double max, long start, long end) {return stringRedisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max,start, end);}/*** 获取集合的元素, 从大到小排序** @param key* @param start* @param end* @return*/public Set<String> zReverseRange(String key, long start, long end) {return stringRedisTemplate.opsForZSet().reverseRange(key, start, end);}public Set<String> zReverseRangeByScore(String key, long min, long max) {return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max);}/*** 获取集合的元素, 从大到小排序, 并返回score值** @param key* @param start* @param end* @return*/public Set<TypedTuple<String>> zReverseRangeWithScores(String key,long start, long end) {return stringRedisTemplate.opsForZSet().reverseRangeWithScores(key, start,end);}/*** 根据Score值查询集合元素, 从大到小排序** @param key* @param min* @param max* @return*/public Set<String> zReverseRangeByScore(String key, double min,double max) {return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max);}/*** 根据Score值查询集合元素, 从大到小排序** @param key* @param min* @param max* @return*/public Set<TypedTuple<String>> zReverseRangeByScoreWithScores(String key, double min, double max) {return stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key,min, max);}/**** @param key* @param min* @param max* @param start* @param end* @return*/public Set<String> zReverseRangeByScore(String key, double min,double max, long start, long end) {return stringRedisTemplate.opsForZSet().reverseRangeByScore(key, min, max,start, end);}/*** 根据score值获取集合元素数量** @param key* @param min* @param max* @return*/public Long zCount(String key, double min, double max) {return stringRedisTemplate.opsForZSet().count(key, min, max);}/*** 获取集合大小** @param key* @return*/public Long zSize(String key) {return stringRedisTemplate.opsForZSet().size(key);}/*** 获取集合大小** @param key* @return*/public Long zZCard(String key) {return stringRedisTemplate.opsForZSet().zCard(key);}/*** 获取集合中value元素的score值** @param key* @param value* @return*/public Double zScore(String key, Object value) {return stringRedisTemplate.opsForZSet().score(key, value);}/*** 移除指定索引位置的成员** @param key* @param start* @param end* @return*/public Long zRemoveRange(String key, long start, long end) {return stringRedisTemplate.opsForZSet().removeRange(key, start, end);}/*** 根据指定的score值的范围来移除成员** @param key* @param min* @param max* @return*/public Long zRemoveRangeByScore(String key, double min, double max) {return stringRedisTemplate.opsForZSet().removeRangeByScore(key, min, max);}/*** 获取key和otherKey的并集并存储在destKey中** @param key* @param otherKey* @param destKey* @return*/public Long zUnionAndStore(String key, String otherKey, String destKey) {return stringRedisTemplate.opsForZSet().unionAndStore(key, otherKey, destKey);}/**** @param key* @param otherKeys* @param destKey* @return*/public Long zUnionAndStore(String key, Collection<String> otherKeys,String destKey) {return stringRedisTemplate.opsForZSet().unionAndStore(key, otherKeys, destKey);}/*** 交集** @param key* @param otherKey* @param destKey* @return*/public Long zIntersectAndStore(String key, String otherKey,String destKey) {return stringRedisTemplate.opsForZSet().intersectAndStore(key, otherKey,destKey);}/*** 交集** @param key* @param otherKeys* @param destKey* @return*/public Long zIntersectAndStore(String key, Collection<String> otherKeys,String destKey) {return stringRedisTemplate.opsForZSet().intersectAndStore(key, otherKeys,destKey);}/**** @param key* @param options* @return*/public Cursor<TypedTuple<String>> zScan(String key, ScanOptions options) {return stringRedisTemplate.opsForZSet().scan(key, options);}/*** 扫描主键，建议使用* @param patten* @return*/public Set<String> scan(String patten){Set<String> keys = stringRedisTemplate.execute((RedisCallback<Set<String>>) connection -> {Set<String> result = new HashSet<>();try (Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder().match(patten).count(10000).build())) {while (cursor.hasNext()) {result.add(new String(cursor.next()));}} catch (IOException e) {e.printStackTrace();}return result;});return  keys;}/*** 管道技术，提高性能* @param type* @param values* @return*/public List<Object> lRightPushPipeline(String type,Collection<String> values){List<Object> results = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {public Object doInRedis(RedisConnection connection) throws DataAccessException {StringRedisConnection stringRedisConn = (StringRedisConnection)connection;//集合转换数组String[] strings = values.toArray(new String[values.size()]);//直接批量发送stringRedisConn.rPush(type, strings);return null;}});return results;}public List<Object> refreshWithPipeline(String future_key,String topic_key,Collection<String> values){List<Object> objects = stringRedisTemplate.executePipelined(new RedisCallback<Object>() {@Nullable@Overridepublic Object doInRedis(RedisConnection redisConnection) throws DataAccessException {StringRedisConnection stringRedisConnection = (StringRedisConnection)redisConnection;String[] strings = values.toArray(new String[values.size()]);stringRedisConnection.rPush(topic_key,strings);stringRedisConnection.zRem(future_key,strings);return null;}});return objects;}}

实现业务

这里的场景是微服务,任务模块也是,所以所有缓存的操作在这个模块做,在使用feign 对外抛出接口,并且数据库也是单独存在
配置文件

server:port: 10001
spring:application:name: schedulecloud:nacos:discovery:server-addr: 192.168.249.132:8848username: nacospassword: nacosconfig:server-addr: 192.168.249.132:8848file-extension: yamlmain:allow-bean-definition-overriding: true
### 上面是bootstrap.yaml的内容,主要配置nacos和服务名
##nacos中的配置信
spring:
# redisredis:host: 192.168.249.132password: 222222port: 6379datasource:driver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://localhost:3306/leadnews_schedule?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=falseusername: rootpassword: 111111
# 设置Mapper接口所对应的XML文件位置，如果你在Mapper接口中有自定义方法，需要进行该配置
mybatis-plus:mapper-locations: classpath*:mapper/*.xml# 设置别名包扫描路径，通过该属性可以给包中的类注册别名type-aliases-package: com.heima.leadnews.schedule.pojos

实现MybatisPlus快速完成低代码框架在这里插入图片描述
因为日志log类设计到多个操作都会写入log,所以之前的log表有乐观锁的版本字段,启动类中添加Mp集合的乐观锁逻辑

@EnableScheduling
@SpringBootApplication
@MapperScan("com.heima.schedule.mapper")
public class ScheduleApplication {public static void main(String[] args) {SpringApplication.run(ScheduleApplication.class,args);}/*** mybatis-plus乐观锁支持* @return*/@Beanpublic MybatisPlusInterceptor optimisticLockerInterceptor(){MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());return interceptor;}}

指定乐观锁版本字段

在taskservice中实现业务,既然是实现延迟队列,那么一定有添加,取消,消费的基本操作

public interface TaskService extends IService<Taskinfo> {
//    返回当任务idpublic Long addTask(Task task);//todo 无法从redis删除数据public boolean cancelTask(long taskid);public Task poll(int type,int priority);}

实现


@Slf4j
@Service
public class TaskServiceImpl  extends ServiceImpl<TaskinfoMapper, Taskinfo> implements TaskService {@AutowiredTaskinfoLogsMapper taskinfoLogsMapper;
@AutowiredCacheService cacheService;/*** 添加任务到数据库中 并且日志也需要传报* @param task* @return*/@Transactional@Overridepublic Long addTask(Task task) {//1.添加任务到数据库中
//          传递过来的是一个dtolong tinfoId= saveSql(task);if (tinfoId!=0) {task.setTaskId(tinfoId);saveCache(task);return tinfoId;}else{throw new RuntimeException("保存任务到数据库失败");}//避免redis和sql不一致}/*** 删除任务并且保证日志的更新* @param taskid  保存的时候日志id任务id设置的同一个值 所以取数据的时候也可以这样* @return*//*** 取消任务* @param taskId* @return*/@Overridepublic boolean cancelTask(long taskId) {boolean flag = false;//删除任务，更新日志Task task = updateDb(taskId,ScheduleConstants.EXECUTED);//删除redis的数据if(task != null){removeTaskFromCache(task);flag = true;}return false;}/*** 消费任务* @param type* @param priority* @return*/@Overridepublic Task poll(int type,int priority) {Task task = null;try {String key = type+"-"+priority;String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);if(StringUtils.isNotBlank(task_json)){task = JSON.parseObject(task_json, Task.class);if (task.getTaskId()!=null){//更新数据库信息System.out.println(task.getTaskId());updateDb(task.getTaskId(),ScheduleConstants.EXECUTED);}else {System.out.println("没有id信息");}}}catch (Exception e){e.printStackTrace();log.error("poll task exception");}return task;}/*** 删除redis中的任务数据* @param task*/private void removeTaskFromCache(Task task) {String key = task.getTaskType()+"-"+task.getPriority();String s = JSON.toJSONString(task.getTaskType()+"-"+task.getPriority()+task);
//这里不做区分因为之前的数据一定是小于现在时间得cacheService.lRemove(ScheduleConstants.TOPIC+key,0,s);cacheService.zRemove(ScheduleConstants.FUTURE+key, s);}/*** 删除任务，更新任务日志状态* @param taskId* @param status* @return*/private Task updateDb(long taskId, int status) {Task task = null;try {//删除任务removeById(taskId);TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);taskinfoLogs.setStatus(status);taskinfoLogsMapper.updateById(taskinfoLogs);task = new Task();BeanUtils.copyProperties(taskinfoLogs,task);task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());}catch (Exception e){log.error("task cancel exception taskid"+taskId);}return task;}/*** redis 操作必须要和数据库操作区分开保持数据一致* 只把将要5分钟后执行的时间加入队列,如果从设置定时时间开始加,* 那么redis中将会保存很多还有一俩天才执行的任务,并且没每分钟做数据刷新的时候资源消耗变大* @param task* @return*/public void saveCache(Task task){//        获取5分钟后的时间Calendar calendar = Calendar.getInstance();calendar.add(Calendar.MINUTE,5);long furtime = calendar.getTime().getTime();//添加任务到redisif (task.getExecuteTime()<=System.currentTimeMillis()){//   2.1 如果任务时间小于等于当前时间存入list 立即执行String key=ScheduleConstants.TOPIC+ task.getTaskType()+"-"+task.getPriority();cacheService.lLeftPush(key, JSON.toJSONString(task));}//2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中else if(task.getExecuteTime()>System.currentTimeMillis()&&task.getExecuteTime()<=furtime){String key=ScheduleConstants.FUTURE+ task.getTaskType()+"-"+task.getPriority();cacheService.zAdd( key, JSON.toJSONString(task), task.getExecuteTime());}}public long saveSql(Task task){//1.添加任务到数据库中
//          传递过来的是一个dtoTaskinfo taskinfo = taskTransinfo(task);long takeId=0;try{save(taskinfo);//1.1 对日志进行保存takeId=taskinfo.getTaskId();TaskinfoLogs logs = InitaskLogs(taskinfo);taskinfoLogsMapper.insert(logs);return takeId;}catch (Exception e){throw new IllegalStateException("保存任务到数据库失败");}}//dto 和pojo 之间有些属性不能bean转换public Taskinfo taskTransinfo(Task task) {Taskinfo taskinfo = new Taskinfo();BeanUtils.copyProperties(task, taskinfo);//info对象中的执行时间是date 这里和long 进行转换taskinfo.setExecuteTime(new Date(task.getExecuteTime()));return taskinfo;}/*** 日志进行初始化* @param taskinfo* @return*/public  TaskinfoLogs InitaskLogs(Taskinfo taskinfo) {TaskinfoLogs taskinfoLogs = new TaskinfoLogs();BeanUtils.copyProperties(taskinfo, taskinfoLogs);
//        初始化乐观锁的版本号taskinfoLogs.setVersion(1);taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);return taskinfoLogs;}/*** 数据同步刷新* 每隔一分钟坚持数据是否快到执行时间*//*** 未来数据定时刷新*/@Scheduled(cron = "0 */1 * * * ?")public void refresh(){String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);if(StringUtils.isNotBlank(token)){log.info("未来数据定时刷新---定时任务");//获取所有未来数据的集合keySet<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");for (String futureKey : futureKeys) {//future_100_50//获取当前数据的key  topicString topicKey = ScheduleConstants.TOPIC+futureKey.split(ScheduleConstants.FUTURE)[1];//按照key和分值查询符合条件的数据Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());//同步数据if(!tasks.isEmpty()){cacheService.refreshWithPipeline(futureKey,topicKey,tasks);log.info("成功的将"+futureKey+"刷新到了"+topicKey);}}}}/*** 将数据库中到到期的未来任务同步一遍*/@Scheduled(cron = "0 */5 * * * ?")@PostConstruct//和启动类初始化同时执行public void refreshData() {clearCache();Calendar calendar=Calendar.getInstance();calendar.add(Calendar.MINUTE,5);List<Taskinfo> taskinfoList = lambdaQuery().lt(Taskinfo::getExecuteTime, calendar.getTimeInMillis()).list();
//        数据的任务添加到redisif(taskinfoList.size()>0&& taskinfoList!=null){taskinfoList.forEach(i->{Task task = new Task();BeanUtils.copyProperties(i,task);task.setExecuteTime(i.getExecuteTime().getTime());//保存到保证数据库和redis的一个同步addTask(task);});}log.info("数据库和redis 进行同步");}public void clearCache(){//        清楚缓存中的数据Set<String> topicKeys = cacheService.scan(ScheduleConstants.TOPIC + "*");Set<String> furtureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");cacheService.delete(topicKeys);cacheService.delete(furtureKeys);}}

逻辑任务添加时候插入数据库,以及任务日志然后根据执行时间和系统设置预处理缓冲时间(这里指的是5分钟),取消任务时,删除任务数据,修改日志状态,其中redis删除数据,并且有定时刷新set队列,将时间满足的移动到list立即执行队列
对外抛出

@RequestMapping("/api/task")
@RestController
public class ScheduleClient implements IScheduleClient {@AutowiredTaskService taskService;@RequestMapping("/add")@Overridepublic ResponseResult addTask(@RequestBody  Task task) {return ResponseResult.okResult(taskService.addTask(task));}@GetMapping("/{taskId}")@Overridepublic ResponseResult cancelTask(@PathVariable("taskId") long taskid) {return ResponseResult.okResult(taskService.cancelTask(taskid));}@GetMapping("/{type}/{priority}")@Overridepublic ResponseResult poll( @PathVariable("type") int type,@PathVariable("priority") int priority) {return ResponseResult.okResult(taskService.poll(type, priority));}
}

feign模块添加对应调用者

@FeignClient(value = "schedule")
public interface IScheduleClient {/**** @param task* @return 任务id*/@PostMapping("/api/v1/task/add")public ResponseResult addTask(@RequestBody Task task);/**** @param taskid* @return 是否成功*/@GetMapping("/api/v1/task/{taskId}")public ResponseResult cancelTask(@PathVariable("taskId") long taskid);@GetMapping("/api/v1/task/{type}/{priority}")public ResponseResult poll( @PathVariable("type") int type,@PathVariable("priority") int priority);
}

发布文章的服务在完成文章的处理逻辑后，调用该模块的添加任务方法,根据执行时间放在哪一个队列,set/list,其中值得主义的是,数据表之间的参数字段是长比特类型,Mp映射也是,所以在其他模块需要调用该缓存模块方法时候,对传递的任务擦拭布参数进行字节序列化化在这里插入图片描述调用者模块

 @Override@Asyncpublic void addNewsToTask(Integer Newsid, Date published) {log.info("addNewsToTask Newsid:"+Newsid+" published:"+published);if (published == null||Newsid==null){throw new IllegalArgumentException("传递参数不全");}Task task = new Task();task.setExecuteTime(published.getTime());task.setTaskType(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType());task.setPriority(TaskTypeEnum.NEWS_SCAN_TIME.getPriority());WmNews news = new WmNews();news.setId(Newsid);task.setParameters(ProtostuffUtil.serialize(news));schduleClient.addTask(task);log.info("addNewsToTask success");}

在调用feign api之前需要把对应的参数准备,其中包括序列化,而这里的序列化为bite采用的是第三方库
序列化工具对比

JdkSerialize：java内置的序列化能将实现了Serilazable接口的对象进行序列化和反序列化， ObjectOutputStream的writeObject()方法可序列化对象生成字节数组
Protostuff：google开源的protostuff采用更为紧凑的二进制数组，表现更加优异，然后使用protostuff的编译工具生成pojo类
所以这里采用protostuff 库

<dependency><groupId>io.protostuff</groupId><artifactId>protostuff-core</artifactId></dependency><dependency><groupId>io.protostuff</groupId><artifactId>protostuff-runtime</artifactId></dependency>