elastic-job 分布式定時任務框架 在 SpringBoot 中如何使用（二）動態添加任務需求

之前一篇用過了如何在使用創建最簡單的任務：比如每天定時清空系統的緩存

這篇文章主要講解：如何運用elastic-job-lite做靈活的細粒度任務，比如：

如何定時取消某個訂單在下訂單后30分鍾未支付的訂單，並改變訂單狀態？

如何讓某個用戶在獲得7天體驗會員在七天后改變這個會員的會員狀態？

某個用戶想定時發布一篇文章？

如何給某個會員在生日當天發送一條祝福短信？

 

elastic-job-lite 就能實現這樣的需求……

 主要是任務配置，任務執行類都是一樣的，下面貼出了demo，僅限於單應用節點時，主要為了實現如何動態的配置任務參數並達到上述需求，方法應用比較簡單

 首先要有任務（作業）類，並交給spring管理類

/*
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package com.dianji.task_server.job.exec;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.MessageFormat;

@Slf4j
@Component
public class OrderExpireJob implements SimpleJob {
    @Value("${serverFlag}")
    private String serverFlag;

    @Override
    public void execute(final ShardingContext shardingContext) {
        int shardingItem = shardingContext.getShardingItem();
        String jobName = shardingContext.getJobName();
        String jobParameter = shardingContext.getJobParameter();
        String logRule = "「執行訂單超時任務」任務名：{0},訂單號:{1},任務分片索引：{2},服務進程「{3}」";
        String logStr = MessageFormat.format(logRule, jobName, jobParameter, shardingItem, serverFlag);
        log.info(logStr);
    }
}

 

接着就是任務（作業）配置了

package com.dianji.task_server.job.config;

import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.dianji.task_server.job.exec.OrderExpireJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 動態添加任務配置
 *
 * @author [e-mail]
 * @create 2018-10-25 下午5:16
 **/
@Slf4j
@Component
public class DynamicAddJobConfig {
    @Autowired
    private ZookeeperRegistryCenter regCenter;

    @Autowired
    private JobEventConfiguration jobEventConfiguration;

    public void dynamicAddSimpleJobScheduler(SimpleJob simpleJob, String jobName, String jobParameter, String cron,
                                             int shardingTotalCount, String shardingItemParameters) {
        new SpringJobScheduler(
                simpleJob,
                regCenter,
                getLiteJobConfiguration(
                        jobName,
                        jobParameter,
                        OrderExpireJob.class,
                        cron,
                        shardingTotalCount,
                        shardingItemParameters),
                jobEventConfiguration).init();
    }

    /**
     * 任務配置
     *
     * @param jobName
     * @param jobParameter
     * @param jobClass
     * @param cron
     * @param shardingTotalCount
     * @param shardingItemParameters
     * @return
     */
    private LiteJobConfiguration getLiteJobConfiguration(
            final String jobName,
            final String jobParameter,
            final Class<? extends SimpleJob> jobClass,
            final String cron,
            final int shardingTotalCount,
            final String shardingItemParameters) {
        return LiteJobConfiguration.newBuilder(
                new SimpleJobConfiguration(
                        JobCoreConfiguration.newBuilder(
                                jobName,
                                cron,
                                shardingTotalCount
                        ).shardingItemParameters(shardingItemParameters).jobParameter(jobParameter).build(),
                        jobClass.getCanonicalName()
                )
        ).overwrite(true).build();
    }
}

 

最后，主動觸發任務添加，這里用了一個restful API 的URL來請求 測試 「讓某個訂單1分鍾后執行過期作業」  任務添加

package com.dianji.task_server.web.controller;

import com.dianji.task_server.job.config.DynamicAddJobConfig;
import com.dianji.task_server.job.exec.OrderExpireJob;
import com.dianji.task_server.util.ResultUtils;
import com.dianji.task_server.web.vo.Result;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

/**
 * 測試控制器
 *
 * @author [e-mail]
 * @create 2018-10-17 上午9:46
 **/
@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {
    @Autowired
    DynamicAddJobConfig dynamicAddJobConfig;

    @Autowired
    private OrderExpireJob orderExpireJob;

    @GetMapping("/addTask")
    public Result addTask(HttpServletRequest request) {
        Date date = new Date(); // 當前時間
        String orderNo = String.valueOf(date.hashCode()); // 訂單號
        String jobName = "OrderExpireJob-" + orderNo; // 任務名稱（不能重復，不然容易覆蓋掉）
        Date expireTime = addMin(date, 1); // 測試時，1分鍾即可
        String cron = testGetCron(expireTime); // 得到cron表達式
        String jobParameter = orderNo; // 將訂單號作為參數
        int shardingTotalCount = 1; // 分片總數
        String shardingItemParameters = "0=a"; // 分片參數
        dynamicAddJobConfig.dynamicAddSimpleJobScheduler(orderExpireJob, jobName, jobParameter, cron,
                shardingTotalCount, shardingItemParameters);
        log.info("「添加訂單超時任務」，任務名{},訂單號{}", jobName, jobParameter);
        return ResultUtils.success();
    }

    /**
     * 僅測試使用方法，日期轉cron表達式
     *
     * @param date 待處理日期
     * @return
     */
    private String testGetCron(java.util.Date date) {
        String dateFormat = "ss mm HH dd MM ? yyyy";
        SimpleDateFormat sdf = new SimpleDateFormat(dateFormat);
        String formatTimeStr = "";
        if (date != null) {
            formatTimeStr = sdf.format(date);
        }
        return formatTimeStr;
    }

    /**
     * 僅測試使用方法，給指定的日期添加分鍾
     *
     * @param oldDate 需要處理日期
     * @param number  添加的分鍾數
     * @return
     */
    private Date addMin(Date oldDate, int number) {

        Calendar c = Calendar.getInstance();
        c.setTime(oldDate);
        c.add(Calendar.MINUTE, number);// 添加分鍾

        return c.getTime();
    }
}

 

應用跑起來后，訪問 /test/addTask 查看日志結果：

 

 

 

作業維護后台，能看見執行的這些 訂單過期任務