前言：

最近在做一个基于SpringCloud+Springboot+Docker的新闻头条微服务项目，用的是黑马的教程，现在项目开发进入了尾声，我打算通过写文章的形式进行梳理一遍，并且会将梳理过程中发现的Bug进行修复，有需要改进的地方我也会继续做出改进。这一系列的文章我将会放入微服务项目专栏中，这个项目适合刚接触微服务的人作为练手项目，假如你对这个项目感兴趣你可以订阅我的专栏进行查看，需要资料可以私信我，当然要是能给我点个小小的关注就更好了，你们的支持是我最大的动力。

一：前期准备

一：前期准备

1.需求分析

当创作者创作好文章之后可以选择立马发布，还能选择定时发布（见下图），这个相信大家在CSDN创作时候都知道，我们需要使用延迟任务来实现文章的定时发布。

编辑

2. 延迟任务概述

在介绍延迟任务之前，我们先了解一下定时任务，定时任务就是有固定的的执行频率，每隔一段时间就执行一次，定时任务与延迟任务区别如下：

定时任务：有固定周期的，有明确的触发时间
延迟任务：没有固定的开始时间，它常常是由一个事件触发的，而在这个事件触发之后的一段时间内触发另一个事件，任务可以立即执行，也可以延迟。
延迟任务使用场景：

商品下单30分钟之内没有付款则将订单取消
接口对接出现网络问题，1分钟之后重试，如果再失败则2分钟之后重试，直至达到阈值

3.技术对比

DelayQueue
JDK自带DelayQueue 是一个支持延时获取元素的阻塞队列，内部采用优先队列 PriorityQueue 存储元素，同时元素必须实现 Delayed 接口；在创建元素时可以指定多久才可以从队列中获取当前元素，只有在延迟期满时才能从队列中提取元素。 DelayQueue属于排序队列，它的特殊之处在于队列的元素必须实现Delayed接口，该接口需要实现compareTo和getDelay方法
getDelay方法：获取元素在队列中的剩余时间，只有当剩余时间为0时元素才可以出队列。
compareTo方法：用于排序，确定元素出队列的顺序。

需要注意的是，使用线程池或者原生DelayQueue程序挂掉之后，任务都是放在内存，需要考虑未处理消息的丢失带来的影响，要保证数据不丢失，就需要持久化（磁盘）。

RabbitMQ实现延迟任务

RabbitMQ实现延迟任务有两种形式，一种是传统的TTL+DLX（死信交换机）来实现，另一种是直接使用插件。TTL+DLX的实现原理是给每个队列设置过期时间，当消息过期之后变成Dead message，就将死信消息发送到另一个交换机，这个交换机叫做死信交换机。

不过更方便的还是直接使用RabbitMQ提供的延迟队列插件比较方便，实践过程可以翻看我前面关于微信支付的文章，里面就是使用了RabbitMQ提供的延迟插件来实现订单的管理。

Redis实现

由于项目后面还需要用到Redis缓存热度靠前的文章，所以这里我选择了使用Redis来实现延迟队列，也正好可以了解其实现原理。

我们都知道Redis中的ZSet数据类型可以实现对数据的排序，那么我们就可以利用这个特点来实现延迟队列，使用时间戳作为score来进行排序。

4.实现思路

编辑

1.为什么任务需要存储在数据库中？

延迟任务是一个通用的服务，任何需要延迟得任务都可以调用该服务，需要考虑数据持久化的问题，存储数据库中是一种数据安全的考虑。

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

效率问题，算法的时间复杂度

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

任务模块是一个通用的模块，项目中任何需要延迟队列的地方，都可以调用这个接口，要考虑到数据量的问题，如果数据量特别大，为了防止阻塞，只需要把未来几分钟要执行的数据存入缓存即可。

二：环境搭建

1.搭建模块

①在service模块下创建一个tbug-headlines-schedule模块

编辑

②添加bootstrap.yml

server:
  port: 51701
spring:
  application:
    name: headlines-schedule
  cloud:
    nacos:
      discovery:
        server-addr: 49.234.52.192:8848
      config:
        server-addr: 49.234.52.192:8848
        file-extension: yml

③在nacos添加相关配置

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

2.数据库准备

taskinfo任务表

编辑

taksinfo_log任务日志表

编辑 3.在docker中安装redis

略

4.项目中集成redis

①在项目中导入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>

②在nacos添加redis配置信息

spring:
  redis:
    host: 49.234.52.192
    password: 440983
    port: 6379

③拷贝CacheService类到tbug-headlines-common模块下，并添加自动配置

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  com.my.common.exception.ExceptionCatch,\
  com.my.common.swagger.SwaggerConfiguration,\
  com.my.common.swagger.Swagger2Configuration,\
  com.my.common.tencentcloud.TextDetection,\
  com.my.common.tencentcloud.ImageDetection,\
  com.my.common.tess4j.Tess4jClient,\
  com.my.common.redis.CacheService,\

三：代码编写

1.实体类导入

①创建task类，用于接收添加任务的参数

package com.my.model.schedule.dtos;
import lombok.Data;
import java.io.Serializable;
@Data
public class Task implements Serializable {
    /**
     * 任务id
     */
    private Long taskId;
    /**
     * 类型
     */
    private Integer taskType;
    /**
     * 优先级
     */
    private Integer priority;
    /**
     * 执行id
     */
    private long executeTime;
    /**
     * task参数
     */
    private byte[] parameters;
}

②创建mapper

package com.my.schedule.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.my.model.schedule.pojos.Taskinfo;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.util.Date;
import java.util.List;
/**
 * <p>
 *  Mapper 接口
 * </p>
 *
 * @author itheima
 */
@Mapper
public interface TaskInfoMapper extends BaseMapper<Taskinfo> {
    List<Taskinfo> queryFutureTime(@Param("taskType")int type, @Param("priority")int priority, @Param("future")Date future);
}

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.my.schedule.mapper.TaskInfoMapper">
    <select id="queryFutureTime" resultType="com.my.model.schedule.pojos.Taskinfo">
        select *
        from taskinfo
        where task_type = #{taskType}
          and priority = #{priority}
          and execute_time <![CDATA[<]]> #{future,javaType=java.util.Date}
    </select>
</mapper>

package com.my.schedule.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.my.model.schedule.pojos.TaskinfoLogs;
import org.apache.ibatis.annotations.Mapper;
/**
 * <p>
 *  Mapper 接口
 * </p>
 *
 * @author itheima
 */
@Mapper
public interface TaskInfoLogsMapper extends BaseMapper<TaskinfoLogs> {
}

2.创建taskService

package com.my.schedule.service;
import com.my.model.schedule.dtos.Task;
/**
 * 对外访问接口
 */
public interface TaskService {
    /**
     * 添加任务接口
     * @param task  任务对象
     * @return  任务ID
     */
    long addTask(Task task);
    /**
     * 取消任务
     * @param taskId        任务id
     * @return              取消结果
     */
    boolean cancelTask(long taskId);
    /**
     * 按照类型和优先级来拉取任务
     * @param type
     * @param priority
     * @return
     */
    Task poll(int type,int priority);
}

3.功能实现

ScheduleConstants常量类

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

TaskServiceImpl

package com.my.schedule.service.serviceImpl;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.my.common.constans.ScheduleConstants;
import com.my.common.redis.CacheService;
import com.my.model.schedule.dtos.Task;
import com.my.model.schedule.pojos.Taskinfo;
import com.my.model.schedule.pojos.TaskinfoLogs;
import com.my.schedule.mapper.TaskInfoLogsMapper;
import com.my.schedule.mapper.TaskInfoMapper;
import com.my.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;
@Slf4j
@Service
@Transactional
public class TaskServiceImpl implements TaskService {
    /**
     * 添加延迟任务
     * @param task  任务对象
     * @return
     */
    @Override
    public long addTask(Task task) {
        //1.将任务添加到数据库中
        boolean success = addTaskToDb(task);
        if(success) {
            //2.将任务添加到redis
            addTaskToCache(task);
        }
        return task.getTaskId();
    }
    @Autowired
    private TaskInfoMapper taskInfoMapper;
    @Autowired
    private TaskInfoLogsMapper taskInfoLogsMapper;
    /**
     * 将任务添加到数据库中
     * @param task
     * @return
     */
    private boolean addTaskToDb(Task task) {
        boolean flag = false;
        try {
            //1.保存任务表
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskInfoMapper.insert(taskinfo);
            //2.设置任务id
            task.setTaskId(taskinfo.getTaskId());
            //3.保存任务日志数据
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo,taskinfoLogs);
            taskinfoLogs.setVersion(1);
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskInfoLogsMapper.insert(taskinfoLogs);
            flag = true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return flag;
    }
    @Autowired
    private CacheService cacheService;
    /**
     * 将任务添加到redis
     * @param task
     */
    private void addTaskToCache(Task task) {
        //1.构造key
        String key = task.getTaskType()+"_"+task.getPriority();
        //2.获取5分钟之后的时间 毫秒级
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);
        long nextScheduleTime = calendar.getTimeInMillis();
        //3.如果任务执行时间小于等于当前时间，存入list
        if(task.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        } else if(task.getExecuteTime() <= nextScheduleTime) {
            //4.如果任务执行时间在5分钟之内，存入zSet
            cacheService.zAdd(ScheduleConstants.FUTURE + key,JSON.toJSONString(task),task.getExecuteTime());
        }
    }
    /**
     * 删除任务
     * @param taskId      任务id
     * @return
     */
    @Override
    public boolean cancelTask(long taskId) {
        boolean flag = false;
        //1.删除任务并更新日志
        Task task = UpdateDb(taskId,ScheduleConstants.EXECUTED);
        //2.从Redis中删除任务
        if(task != null) {
            removeFromCache(task);
            log.info("删除Redis中的任务成功:{}",taskId);
            flag = true;
        }
        return flag;
    }
    /**
     * 从redis中删除任务
     * @param task
     */
    private void removeFromCache(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        if(task.getExecuteTime() <= System.currentTimeMillis()) {
            log.info("删除正要执行的任务...");
            cacheService.lRemove(ScheduleConstants.TOPIC + key,0,JSON.toJSONString(task));
        } else {
            log.info("删除将要执行的任务...");
            cacheService.zRemove(ScheduleConstants.FUTURE + key,JSON.toJSONString(task));
        }
    }
    /**
     * 删除任务，更新日志
     * @param taskId
     * @param status
     * @return
     */
    private Task UpdateDb(long taskId, int status) {
        Task task = null;
        try {
            //1.删除任务
            log.info("删除数据库中的任务...");
            taskInfoMapper.deleteById(taskId);
            //2.更新日志
            log.info("更新任务日志...");
            TaskinfoLogs taskinfoLogs = taskInfoLogsMapper.selectById(taskId);
            taskinfoLogs.setStatus(status);
            taskInfoLogsMapper.updateById(taskinfoLogs);
            //3.设置返回值
            task = new Task();
            BeanUtils.copyProperties(taskinfoLogs,task);
            task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        } catch (BeansException e) {
            throw new RuntimeException(e);
        }
        return task;
    }
    /**
     * 消费任务
     * @param type  任务类型
     * @param priority 任务优先级
     * @return  Task
     */
    @Override
    public 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);
                //更新数据库
                UpdateDb(task.getTaskId(),ScheduleConstants.EXECUTED);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("poll task exception");
        }
        return task;
    }
}

主要包括添加任务、取消任务、消费任务三个功能。

下篇预告：实现数据定时刷新

【Spring Cloud】新闻头条微服务项目：使用Reids延迟队列实现文章定时发布（上）

一：前期准备

1.需求分析

2. 延迟任务概述

3.技术对比

4.实现思路

二：环境搭建

1.搭建模块

2.数据库准备

编辑 3.在docker中安装redis

4.项目中集成redis

三：代码编写

1.实体类导入

2.创建taskService

3.功能实现

微服务

热门文章

最新文章

相关课程

相关电子书

相关实验场景