public class DelayedTask  implements Delayed{
    
    // 任务的执行时间
    private int executeTime = 0;
    
    public DelayedTask(int delay){
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.SECOND,delay);
        this.executeTime = (int)(calendar.getTimeInMillis() /1000 );
    }

    /**
     * 元素在队列中的剩余时间
     * @param unit
     * @return
     */
    @Override
    public long getDelay(TimeUnit unit) {
        Calendar calendar = Calendar.getInstance();
        return executeTime - (calendar.getTimeInMillis()/1000);
    }

    /**
     * 元素排序
     * @param o
     * @return
     */
    @Override
    public int compareTo(Delayed o) {
        long val = this.getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
        return val == 0 ? 0 : ( val < 0 ? -1: 1 );
    }


    public static void main(String[] args) {
        DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>();
        
        queue.add(new DelayedTask(5));
        queue.add(new DelayedTask(10));
        queue.add(new DelayedTask(15));

        System.out.println(System.currentTimeMillis()/1000+" start consume ");
        while(queue.size() != 0){
            DelayedTask delayedTask = queue.poll();
            if(delayedTask !=null ){
                System.out.println(System.currentTimeMillis()/1000+" cosume task");
            }
            //每隔一秒消费一次
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }     
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.
• 54.
• 55.
• 56.

server:
  port: 51701
spring:
  application:
    name: leadnews-schedule
  cloud:
    nacos:
      discovery:
        server-addr: 192.168.141.102:8848
      # nacos热配置中心
      config:
        server-addr: 192.168.141.102:8848
        file-extension: yml

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.

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

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.

package cn.whu.model.schedule.pojos;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;

@Data
@TableName("taskinfo")
public class Taskinfo implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 任务id
     */
    @TableId(type = IdType.ID_WORKER)
    private Long taskId;

    /**
     * 执行时间
     */
    @TableField("execute_time")
    private Date executeTime;

    /**
     * 参数
     */
    @TableField("parameters")
    private byte[] parameters;

    /**
     * 优先级
     */
    @TableField("priority")
    private Integer priority;

    /**
     * 任务类型
     */
    @TableField("task_type")
    private Integer taskType;
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.

package cn.whu.model.schedule.pojos;

import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;


@Data
@TableName("taskinfo_logs")
public class TaskinfoLogs implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 任务id
     */
    @TableId(type = IdType.ID_WORKER)
    private Long taskId;

    /**
     * 执行时间
     */
    @TableField("execute_time")
    private Date executeTime;

    /**
     * 参数
     */
    @TableField("parameters")
    private byte[] parameters;

    /**
     * 优先级
     */
    @TableField("priority")
    private Integer priority;

    /**
     * 任务类型
     */
    @TableField("task_type")
    private Integer taskType;

    /**
     * 版本号,用乐观锁
     */
    @Version
    private Integer version;

    /**
     * 状态 0=int 1=EXECUTED 2=CANCELLED
     */
    @TableField("status")
    private Integer status;

}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.
• 54.
• 55.
• 56.
• 57.
• 58.

/**
* mybatis-plus乐观锁支持
* @return
*/
@Bean
public MybatisPlusInterceptor optimisticLockerInterceptor(){
    MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
    interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());
    return interceptor;
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.

docker pull redis

• 1.

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

• 1.

<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>

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.

spring:
  redis:
    host: 192.168.141.102
    password: leadnews
    port: 6379

• 1.
• 2.
• 3.
• 4.
• 5.

package cn.whu.schedule.test;

import cn.whu.common.redis.CacheService;
import cn.whu.schedule.ScheduleApplication;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.util.Set;

@SpringBootTest(classes = ScheduleApplication.class)
@RunWith(SpringRunner.class)
public class RedisTest {

    @Resource
    private CacheService cacheService;

    @Test
    public void testList1(){
        // 在list的左边添加元素
        cacheService.lLeftPush("list_001","hello,redis1");//点进去看一下api的封装就是往list头部插入
        cacheService.lLeftPush("list_001","hello,redis2");//点进去看一下api的封装就是往list头部插入
    }

    @Test
    public void testList2(){
        // 在list的右边获取元素，并删除
        String list001 = cacheService.lRightPop("list_001");
        System.out.println(list001); // hello,redis1
    }


    @Test
    public void testZset1(){
        // 添加数据到zset中 有分值的
        cacheService.zAdd("zset_key_001","hello zset 001",1000);
        cacheService.zAdd("zset_key_001","hello zset 002",8888);
        cacheService.zAdd("zset_key_001","hello zset 003",7777);
        cacheService.zAdd("zset_key_001","hello zset 004",9999);
    }

    @Test
    public void testZset2(){
        // 按照分值获取数据
        // 获取分值在0~8888内的元素，且(应该自动是)按照分值升序排列
        Set<String> zsetKey001 = cacheService.zRangeByScore("zset_key_001", 0, 8888);
        System.out.println(zsetKey001);
        // [hello zset 001, hello zset 003, hello zset 002]
    }

}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.

package cn.whu.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;
    
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.

package cn.whu.schedule.service;

import cn.whu.model.schedule.dtos.Task;

/**
 * 对外访问接口
 */
public interface TaskService {

    /**
     * 添加任务
     * @param task   任务对象
     * @return       任务id
     */
    public long addTask(Task task) ;

}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.

package cn.whu.schedule.service.impl;

import cn.whu.common.constants.ScheduleConstants;
import cn.whu.common.redis.CacheService;
import cn.whu.model.schedule.dtos.Task;
import cn.whu.model.schedule.pojos.Taskinfo;
import cn.whu.model.schedule.pojos.TaskinfoLogs;
import com.alibaba.fastjson.JSON;
import cn.whu.schedule.mapper.TaskinfoLogsMapper;
import cn.whu.schedule.mapper.TaskinfoMapper;
import cn.whu.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;

@Service
@Slf4j
@Transactional
public class TaskServiceImpl implements TaskService {

    // 访问db的两个mapper
    @Resource
    private TaskinfoMapper taskinfoMapper;

    @Resource
    private TaskinfoLogsMapper taskinfoLogsMapper;

    // 操作redis
    @Resource
    private CacheService cacheService;

    /**
     * 添加任务
     *
     * @param task 任务对象
     * @return 任务id
     */
    @Override
    public long addTask(Task task) {
        // 1. 添加任务到数据库中
        boolean success = addTaskToDb(task);
        if (!success) return -1; // 写db失败，直接返回-1失败

        // 2. 添加任务到redis中
        addTaskToCache(task);

        return task.getTaskId();
    }

    private void addTaskToCache(Task task) {
        // 任务类型+优先级 就可以唯一标定一个task
        String key = task.getTaskType() + "_" + task.getPriority();
        long delay = 5 * 60 * 1000;//延迟时间 5min

        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            // 2.1 如果任务的执行时间<=当前时间，存入list (redis的list结构)
            cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        } else if (task.getExecuteTime() <= System.currentTimeMillis() + delay) {
            // 2.2 如果任务的执行时间>当前时间 && 小于等于预设时间(未来5分钟) 存入zset (redis的zset结构)
            cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
            // 分值就是任务执行时间的ms值
        }
    }

    /**
     * 添加任务到数据库中
     *
     * @param task
     * @return
     */
    private boolean addTaskToDb(Task task) {
        boolean flag = true;

        try {
            // 1. 保存任务表
            // 1.1 准备数据
            Taskinfo taskinfo = new Taskinfo();
            // 1)拷贝数据
            BeanUtils.copyProperties(task, taskinfo);
            // 2)特殊字段处理：执行时间的类型不一样，long->Date, 需要手动处理
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            // 1.2 写db
            taskinfoMapper.insert(taskinfo);

            // 设置一下taskId  task引用传递，可以返回到主调方
            task.setTaskId(taskinfo.getTaskId());

            // 2. 保存任务日志数据
            // 2.1 准备数据
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            // 1)拷贝数据
            BeanUtils.copyProperties(taskinfo, taskinfoLogs);
            // 2)特殊字段处理
            taskinfoLogs.setVersion(1); // 乐观锁版本号
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED); // 初始化(init)状态0
            // 2.2 写DB
            taskinfoLogsMapper.insert(taskinfoLogs);
        } catch (BeansException e) {
            flag = false;
            log.info("TaskServiceImpl-addTaskToDb exception task.id:{}", task.getTaskId(), e);
            e.printStackTrace();
        }

        return flag;
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.
• 54.
• 55.
• 56.
• 57.
• 58.
• 59.
• 60.
• 61.
• 62.
• 63.
• 64.
• 65.
• 66.
• 67.
• 68.
• 69.
• 70.
• 71.
• 72.
• 73.
• 74.
• 75.
• 76.
• 77.
• 78.
• 79.
• 80.
• 81.
• 82.
• 83.
• 84.
• 85.
• 86.
• 87.
• 88.
• 89.
• 90.
• 91.
• 92.
• 93.
• 94.
• 95.
• 96.
• 97.
• 98.
• 99.
• 100.
• 101.
• 102.
• 103.
• 104.
• 105.
• 106.
• 107.
• 108.
• 109.
• 110.
• 111.

package cn.whu.common.constants;

public class ScheduleConstants {

    //task状态
    public static final int SCHEDULED=0;   //初始化状态

    public static final int EXECUTED=1;       //已执行状态

    public static final int CANCELLED=2;   //已取消状态

    public static String FUTURE="future_";   //未来数据key前缀

    public static String TOPIC="topic_";     //当前数据key前缀
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.

package cn.whu.schedule.service.impl;

import cn.whu.model.schedule.dtos.Task;
import cn.whu.schedule.ScheduleApplication;
import cn.whu.schedule.service.TaskService;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;

@SpringBootTest(classes = ScheduleApplication.class)
@RunWith(SpringRunner.class)
class TaskServiceImplTest {

    @Resource
    private TaskService taskService;

    @Test
    void addTask() {
        Task task = new Task();
        task.setTaskType(100);
        task.setPriority(50);
        task.setParameters("task test".getBytes());
        task.setExecuteTime(System.currentTimeMillis());

        long taskId = taskService.addTask(task);
        System.out.println("taskId = " + taskId);

    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.

@SpringBootTest(classes = ScheduleApplication.class)
@RunWith(SpringRunner.class)
class TaskServiceImplTest {

    @Resource
    private TaskService taskService;

    @Test
    void addTask() {
        Task task = new Task();
        task.setTaskType(100);
        task.setPriority(50);
        task.setParameters("task test".getBytes());
        //task.setExecuteTime(System.currentTimeMillis());
        task.setExecuteTime(System.currentTimeMillis()+500);

        long taskId = taskService.addTask(task);
        System.out.println("taskId = " + taskId);

    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.

task.setExecuteTime(System.currentTimeMillis()+500000); // 超过5分钟 redis中就没有新记录了

• 1.

/**
 * 取消任务
 * @param task 任务对象
 * @return 取消成功还是失败
 */
public boolean cancelTask(long taskId);

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.

/*---------------------------删除任务-----------------------------*/

/**
 * 取消任务
 *
 * @param taskId
 * @return 取消成功还是失败
 */
@Override
public boolean cancelTask(long taskId) {
    boolean flag = false;

    // 删除任务，更新任务日志 (taskinfo表删除一条记录  taskinfo_logs表更新一条记录)
    Task task = updateDb(taskId, ScheduleConstants.CANCELLED);
    // 删除redis需要task的两个字段找到key，以及执行时间判断在哪里
    // 所以更新完db干脆直接返回task

    // 删除redis的数据(任务记录)
    if (task != null) {
        removeTaskFromCache(task);
        flag = true;
    }

    return flag;
}

/**
 * 删除redis中的数据 (就是那条任务记录)
 *
 * @param task
 */
private void removeTaskFromCache(Task task) {
    // 任务类型+优先级  确定一个任务队列 list
    String key = task.getTaskType() + "_" + task.getPriority();

    if (task.getExecuteTime() <= System.currentTimeMillis()) {
        // 删除根据value来删的 key只是找到了那个任务队列
        cacheService.lRemove(ScheduleConstants.TOPIC + key, 0, JSON.toJSONString(task));
    } else {
        // 注意这里没有index参数了 (因为set不会重复的)
        cacheService.zRemove(ScheduleConstants.FUTURE + key, JSON.toJSONString(task));
    }
}

private Task updateDb(long taskId, int status) {
    Task task = null;
    try {
        // 删除db表-taskinfo 记录
        taskinfoMapper.deleteById(taskId);
        // 更新db表-taskinfo_logs 记录
        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
        taskinfoLogs.setStatus(status);
        taskinfoLogsMapper.updateById(taskinfoLogs);
        // 返回刚删除的task数据
        task = new Task();
        BeanUtils.copyProperties(taskinfoLogs, task);
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
    } catch (BeansException e) {
        log.error("task cancel exception taskId={}", taskId);
        e.printStackTrace();
    }
    return task;
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.
• 54.
• 55.
• 56.
• 57.
• 58.
• 59.
• 60.
• 61.
• 62.
• 63.

@SpringBootTest(classes = ScheduleApplication.class)
@RunWith(SpringRunner.class)
class TaskServiceImplTest {
	@Resource
	private TaskService taskService;
	
	@Test
	public void cancelTask(){
	    taskService.cancelTask(1802971828875091969l);
	}
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.

/**
 * 按照类型和优先级拉取任务
 * 类型+优先级 -》 确定任务队列
 * @param type
 * @param priority
 * @return
 */
public Task poll(int type,int priority);

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.

/**
 * 按照类型和优先级拉取任务
 * 类型+优先级 -》 确定任务队列
 *
 * @param type
 * @param priority
 * @return
 */
@Override
public Task poll(int type, int priority) {
    Task task = null;

    try {
        // 1. 从redis拉取数据
        String key = type + "_" + priority;
        // 待消费的任务只能在list中
        String taskJson = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
        if (StringUtils.isNotBlank(taskJson)) {
            task = JSON.parseObject(taskJson, Task.class);

            // 2. 修改db数据
            // 删除任务  日志状态修改为已执行
            updateDb(task.getTaskId(), ScheduleConstants.EXECUTED);
        }
    } catch (Exception e) {
        log.error("TaskServiceImpl.poll task error taskType={},taskPriority={}", type, priority);
        e.printStackTrace();
    }

    return task;
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.

@SpringBootTest(classes = ScheduleApplication.class)
@RunWith(SpringRunner.class)
class TaskServiceImplTest {

    @Resource
    private TaskService taskService;

    @Test
    void addTask() {
        Task task = new Task();
        task.setTaskType(100);
        task.setPriority(50);
        task.setParameters("task test".getBytes());
        task.setExecuteTime(System.currentTimeMillis());
        long taskId = taskService.addTask(task);
        System.out.println("taskId = " + taskId);

    }

    @Test
    public void poll(){
        Task task = taskService.poll(100, 50);
        System.out.println("task = " + task);
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.

task = Task(taskId=1802996859323768834, taskType=100, priority=50, executeTime=1718702923817, parameters=[116, 97, 115, 107, 32, 116, 101, 115, 116])

• 1.

@Test
void addTasks() {
    Task task = new Task();
    task.setTaskType(100);
    task.setPriority(50);
    task.setParameters("task test".getBytes());

    for (int i = 101; i <= 105; i++) {
        task.setTaskType(i);
        task.setExecuteTime(System.currentTimeMillis() + 500);
        task.setTaskId(null);//写db时不能有id
        long taskId = taskService.addTask(task);
        System.out.println("taskId = " + taskId);
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.

@Test
public void testKeys(){
    Set<String> keys = cacheService.keys("future_*");
    System.out.println(keys);

    Set<String> scan = cacheService.scan("future_*"); // 一般情况下都用scan
    System.out.println(scan);
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.

//耗时4864
@Test
public  void testPipe1(){
    long start =System.currentTimeMillis();
    for (int i = 0; i <10000 ; i++) {
        Task task = new Task();
        task.setTaskType(1001);
        task.setPriority(1);
        task.setExecuteTime(new Date().getTime());
        // 只push到list中  redis的一个list中(一个队列中新增1w条数据)
        cacheService.lLeftPush("1001_1", JSON.toJSONString(task));
    }
    System.out.println("耗时"+(System.currentTimeMillis()- start));//耗时4864
}

// 642毫秒
@Test
public void testPipe2(){
    long start  = System.currentTimeMillis();
    //使用管道技术
    List<Object> objectList = cacheService.getstringRedisTemplate().executePipelined(new RedisCallback<Object>() {
        @Nullable
        @Override
        public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
            for (int i = 0; i <10000 ; i++) {
                Task task = new Task();
                task.setTaskType(1001);
                task.setPriority(1);
                task.setExecuteTime(new Date().getTime());
                redisConnection.lPush("1001_1".getBytes(), JSON.toJSONString(task).getBytes());
            }
            return null;
        }
    });
    System.out.println("使用管道技术执行10000次自增操作共耗时:"+(System.currentTimeMillis()-start)+"毫秒");
    // 使用管道技术执行10000次自增操作共耗时:642毫秒
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.

/**
 * 未来数据定时刷新
 *
 * @Scheduled注解就是任务调度注解 括号内容配置的含义就是每分钟执行1次
 * @Scheduled修饰的定时方法必须是无参且无返回值的方法
 */
@Scheduled(cron = "0 */1 * * * ?")
public void refresh() {
    log.info("未来数据定时刷新---定时任务");

    // 获取所有未来数据的keys  (就zset未来任务所有队列名称)
    Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");

    for (String futureKey : futureKeys) {
        // 获取当前任务到list执行队列后的key
        String topicKey = ScheduleConstants.TOPIC + futureKey.substring("future_".length());
        // futureKey: future_100_50
        // topicKey: topic_100_50

        // 按照key和分值查询符合条件的数据
        // 0~当前时间的分数范围内查找  其实就是查(futureKey队列中)小于当前时间的记录
        Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());

        // 同步数据 (futureKey队的数据)
        if (!tasks.isEmpty()) {
            // 将数据tasks，从futureKey，移动到，topicKey
            cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
            log.info("成功地将 {} 刷新到 {}, 本次共刷新 {} 个任务", futureKey, topicKey, tasks.size());
        }
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.

Tue Jun 18 22:44:45 CST 2024 WARN: Establishing SSL connection without server's identity verification is not recommended. According to MySQL 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be established by default if explicit option isn't set. For compliance with existing applications not using SSL the verifyServerCertificate property is set to 'false'. You need either to explicitly disable SSL by setting useSSL=false, or set useSSL=true and provide truststore for server certificate verification.
2024-06-18 22:45:00.024  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 未来数据定时刷新---定时任务
2024-06-18 22:45:00.134  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 成功地将 future_104_50 刷新到 topic_104_50, 本次共刷新 1 个任务
2024-06-18 22:45:00.156  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 成功地将 future_102_50 刷新到 topic_102_50, 本次共刷新 1 个任务
2024-06-18 22:45:00.163  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 成功地将 future_105_50 刷新到 topic_105_50, 本次共刷新 1 个任务
2024-06-18 22:45:00.169  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 成功地将 future_103_50 刷新到 topic_103_50, 本次共刷新 1 个任务
2024-06-18 22:45:00.176  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 成功地将 future_101_50 刷新到 topic_101_50, 本次共刷新 1 个任务
2024-06-18 22:45:00.181  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 成功地将 future_100_50 刷新到 topic_100_50, 本次共刷新 1 个任务
2024-06-18 22:46:00.004  INFO 20188 --- [   scheduling-1] c.h.s.service.impl.TaskServiceImpl       : 未来数据定时刷新---定时任务

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.

/**
 * 加锁
 *
 * @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命令：
        //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;
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.

/**
 * 加锁
 *
 * @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命令：
        //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;
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.

/**
 * 数据库任务定时同步到redis中
 */
@PostConstruct // 微服务启动时会立即执行一次  (防止服务挂掉后重启不能立即同步)
@Scheduled(cron = "0 */5 * * * ?") // 每5分钟执行一次
public void reloadData() {
    // 清理缓存中的数据 list  zset  (db里面重新同步最近的数据到redis 原来的redis缓存可以都不要了)
    clearCache();

    // 查询符合条件的任务
    // 先获取5分钟后的时间实例
    Calendar calendar = Calendar.getInstance();
    calendar.add(Calendar.MINUTE, 5); // calendar类很方便就能实现
    // long ms = calendar.getTimeInMillis(); // ms值
    // Date date = calendar.getTime(); // date日期值
    // 把任务添加到redis
    List<Taskinfo> taskinfoList = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery()
            .lt(Taskinfo::getExecuteTime, calendar.getTime())
    );
    // 把任务添加到redis
    if (taskinfoList != null && taskinfoList.size() > 0) { // !!! 安全性呀
        for (Taskinfo taskinfo : taskinfoList) {
            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(taskinfo.getExecuteTime().getTime());
            addTaskToCache(task);
        }
    }

    log.info("重新清空redis缓存，同步db数据到redis,本次共同步 {} 条数据", taskinfoList.size());

}

/**
 * 清理缓存(redis)中的数据
 */
public void clearCache() {
    Set<String> topicKeys = cacheService.scan(ScheduleConstants.TOPIC + "*");
    Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
    cacheService.delete(topicKeys);
    cacheService.delete(futureKeys);
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.

@Test
void addTasks() {
    Task task = new Task();
    task.setTaskType(100);
    task.setPriority(50);
    task.setParameters("task test".getBytes());

    for (int i = 101; i <= 105; i++) {
        task.setTaskType(i);
        task.setExecuteTime(System.currentTimeMillis() + 500);
        task.setTaskId(null);//写db时不能有id
        long taskId = taskService.addTask(task);
        System.out.println("taskId = " + taskId);
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.

package cn.whu.apis.schedule;


import cn.whu.model.common.dtos.ResponseResult;
import cn.whu.model.schedule.dtos.Task;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;

@FeignClient(value = "leadnews-schedule") // 也可以加个fallback吧
public interface IScheduleClient {

    /**
     * 添加任务
     * @param task   任务对象
     * @return       任务id long
     */
    @PostMapping("/api/v1/task/add")
    public ResponseResult addTask(@RequestBody Task task) ;

    /**
     * 取消任务
     * @param task 任务对象
     * @return 取消成功还是失败 boolean
     */
    @GetMapping("/api/v1/task/{taskId}")
    public ResponseResult cancelTask(@PathVariable("taskId") long taskId);

    /**
     * 按照类型和优先级拉取任务
     * 类型+优先级 -》 确定任务队列
     * @param type
     * @param priority
     * @return Task
     */
    @GetMapping("/api/v1/task/{type}/{priority}")
    public ResponseResult poll(@PathVariable("type") int type,@PathVariable("priority") int priority);

}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.

package cn.whu.schedule.feign;

import cn.whu.apis.schedule.IScheduleClient;
import cn.whu.model.common.dtos.ResponseResult;
import cn.whu.model.schedule.dtos.Task;
import cn.whu.schedule.service.TaskService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;


@RestController
public class ScheduleClient implements IScheduleClient {

    @Autowired
    private TaskService taskService;

    /**
     * 添加任务
     * @param task 任务对象
     * @return 任务id
     */
    @PostMapping("/api/v1/task/add")
    @Override
    public ResponseResult addTask(@RequestBody Task task) {
        return ResponseResult.okResult(taskService.addTask(task));
    }

    /**
     * 取消任务
     * @param taskId 任务id
     * @return 取消结果
     */
    @GetMapping("/api/v1/task/cancel/{taskId}")
    @Override
    public ResponseResult cancelTask(@PathVariable("taskId") long taskId) {
        return ResponseResult.okResult(taskService.cancelTask(taskId));
    }

    /**
     * 按照类型和优先级来拉取任务
     * @param type
     * @param priority
     * @return
     */
    @GetMapping("/api/v1/task/poll/{type}/{priority}")
    @Override
    public ResponseResult poll(@PathVariable("type") int type, @PathVariable("priority") int priority) {
        return ResponseResult.okResult(taskService.poll(type,priority));
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.

package cn.whu.wemedia.service;

import java.util.Date;

public interface WmNewsTaskService {

    /**
     * 添加任务到延迟队列中
     * @param id  文章的id
     * @param publishTime  发布时间  可以作为任务的执行时间
     */
    public void addNewsToTask(Integer id, Date publishTime);
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.

package cn.whu.wemedia.service.impl;

import cn.whu.apis.schedule.IScheduleClient;
import cn.whu.model.common.enums.TaskTypeEnum;
import cn.whu.model.schedule.dtos.Task;
import cn.whu.model.wemedia.pojos.WmNews;
import cn.whu.utils.common.ProtostuffUtil;
import cn.whu.wemedia.service.WmNewsService;
import cn.whu.wemedia.service.WmNewsTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;

@Service
@Slf4j
public class WmNewsTaskServiceImpl implements WmNewsTaskService {

    @Resource
    private IScheduleClient scheduleClient;

    @Resource
    private WmNewsService wmNewsService;

    /**
     * 添加任务到延迟队列中
     *
     * @param id          文章的id
     * @param publishTime 发布时间  可以作为任务的执行时间
     */
    @Override
    @Async
    public void addNewsToTask(Integer id, Date publishTime) {

        log.info("添加任务到到延迟服务中------begin");

        // 1. 封装task
        Task task = new Task();
        task.setExecuteTime(publishTime.getTime());
        task.setTaskType(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType());
        task.setPriority(TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
        // 参数比较麻烦，本来传id就行了，但是需要一个序列化对象
        WmNews wmNews = new WmNews();
        wmNews.setId(id);
        task.setParameters(ProtostuffUtil.serialize(wmNews));

        // 2. feign接口调用定时任务，添加任务到db和redis
        scheduleClient.addTask(task);

        log.info("添加任务到到延迟服务中------end");
    }
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.
• 54.

package cn.whu.model.common.enums;

import lombok.AllArgsConstructor;
import lombok.Getter;

@Getter
@AllArgsConstructor
public enum TaskTypeEnum {

    NEWS_SCAN_TIME(1001, 1,"文章定时审核"),
    REMOTE_ERROR(1002, 2,"第三方接口调用失败，重试");
    private final int taskType; //对应具体业务
    private final int priority; //业务不同级别
    private final String desc; //描述信息
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.

<dependency>
    <groupId>io.protostuff</groupId>
    <artifactId>protostuff-core</artifactId>
    <version>1.6.0</version>
</dependency>

<dependency>
    <groupId>io.protostuff</groupId>
    <artifactId>protostuff-runtime</artifactId>
    <version>1.6.0</version>
</dependency>

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.

/**
 * jdk序列化与protostuff序列化对比
 * @param args
 */
public static void main(String[] args) {
    long start =System.currentTimeMillis();
    for (int i = 0; i <1000000 ; i++) {
        WmNews wmNews =new WmNews();
        JdkSerializeUtil.serialize(wmNews);
    }
    System.out.println(" jdk 花费 "+(System.currentTimeMillis()-start));

    start =System.currentTimeMillis();
    for (int i = 0; i <1000000 ; i++) {
        WmNews wmNews =new WmNews();
        ProtostuffUtil.serialize(wmNews);
    }
    System.out.println(" protostuff 花费 "+(System.currentTimeMillis()-start));
}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.

@Autowired
private WmNewsTaskService wmNewsTaskService;
 
/**
     * 发布修改文章或保存为草稿
     * @param dto
     * @return
     */
@Override
public 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---> string
    if(dto.getImages() != null && dto.getImages().size() > 0){
        //[1dddfsd.jpg,sdlfjldk.jpg]-->   1dddfsd.jpg,sdlfjldk.jpg
        String imageStr = StringUtils.join(dto.getImages(), ",");
        wmNews.setImages(imageStr);
    }
    //如果当前封面类型为自动 -1
    if(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);

    //审核文章
    //        wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
    // 放到db里面，然后再由定时任务根据执行时间慢慢刷新到redis里
    wmNewsTaskService.addNewsToTask(wmNews.getId(),wmNews.getPublishTime());

    return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);

}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.
• 31.
• 32.
• 33.
• 34.
• 35.
• 36.
• 37.
• 38.
• 39.
• 40.
• 41.
• 42.
• 43.
• 44.
• 45.
• 46.
• 47.
• 48.
• 49.
• 50.
• 51.
• 52.
• 53.
• 54.
• 55.

/**
 * 消费延迟队列数据
 */
public void scanNewsByTask();

• 1.
• 2.
• 3.
• 4.

@Autowired
private WmNewsAutoScanServiceImpl wmNewsAutoScanService;

/**
 * 消费延迟队列数据
 * 定时自动执行的任务
 */
@Override
@Scheduled(fixedRate = 1000) // 每秒执行一次
public void scanNewsByTask() {

    log.info("文章审核---消费任务执行---begin---");

    ResponseResult responseResult = scheduleClient.poll(
            TaskTypeEnum.NEWS_SCAN_TIME.getTaskType(),
            TaskTypeEnum.NEWS_SCAN_TIME.getPriority()
    );

    if (responseResult.getCode().equals(200) && responseResult.getData() != null) {
        // responseResult.getData()返回类型是T,强转不合适，用json转
        String jsonString = JSON.toJSONString(responseResult.getData());
        Task task = JSON.parseObject(jsonString, Task.class);
        WmNews wmNews = ProtostuffUtil.deserialize(task.getParameters(), WmNews.class);
        wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
        log.info("文章审核----文章id:{}", wmNews.getId());
    }

    log.info("文章审核---消费任务执行---end---");

}

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.
• 24.
• 25.
• 26.
• 27.
• 28.
• 29.
• 30.