简介

dal-job是一个去中心化的轻量的分布式Job框架。它没有master结点，代码是在各个模块上运行的。

帮助开发人员在分布式环境下开发job时，只用关注业务，而不用去关心job被重复执行的问题。 它主要提供如下功能：

1. 分布式环境下，同一时刻只有一台机器执行job。
2. 记录job的运行日志，包括运行记录，耗时，异常日志等。（job_log）
3. 记录系统下面的job信息。（job_info）
4. 提供内置的重试job，对于需要重试的场景使用。包含：事前重试 和 事后重试
5. dal-job也可以指定多台实例同时运行，但是需要在job的逻辑中保证数据处理不重不漏。
6. 参考方案：取数时可以使用随机生成的全局唯一key，去取数进行处理

问题思考

在分布式环境下，本地去中心化的分布式job需要解决的问题：

1. 我们面对的是可能是一台实例上多线程的环境，需要保证在多线程环境下只有一个线程能执行job。
2. （这种情况比较少，job是受自己模块管理的，一般一台机器一个job只会有一个线程在拉起执行）
3. 我们面对的一定是多进程环境，需要保证在同一时刻多个机器的实例的进程，只有一个进程能执行job
4. 分布式环境下，多个机器实例的系统时间可能不一致，导致job的运行不在同一时刻，我们需要保证这种情况下，只有一台机器对正确执行job

原理

使用数据库的行级锁来保证同一时刻只有一台机在执行任务。

具体：使用 【悲观锁 + JobStatusCheck + TimeLimit】 实现在多线程与多进程（主要是多进程）环境下，一个job在运行过程中，只会有一台机在执行job

技术

使用Quartz + MySQL。同时与Spring友好融合。提供注解（@TimedTask）形式的job配置

Quick Start

定时调度

dal-job支持分布式环境下单台启动 和 多台启动。其中多台启动会在所有的实例上运行，需要自己解决取数问题。

dal-job提供了注解形式的job配置，具体可以参考com.kvn.dal.core.single_node.SingleNodeJob.java

@Target({ TYPE })
@Retention(RUNTIME)
public @interface TimedTask {
 String corn();
 boolean isGlobalSingle() default true; // 分布式环境下，是否单台启动
 String desc() default "";
}

定时调度例子：

@TimedTask(corn = "0 0/1 * * * ?", desc = "测试job222")
@Service
public class MyTestJob2 implements ExecutableTask {
 @Override
 public void execute(JobExecutionContext context) throws JobExecutionException {
 System.out.println(DateTime.now() + "--" + Thread.currentThread().getName() + "---------------doBizJob2222--------");
 try {
			Thread.sleep(3000L);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
 if(new Random().nextInt() % 2 == 0){
 	throw new RuntimeException("biz执行MyTestJob2异常，xxxxxxxxx");
 }
 }
}

重试调度

dal-job提供了内置的重试调度实现，可以方便的对异常数据进行定时重试。

重试分为两种：一是，事前重试；二是，事后重试

事前重试

事前重试，即不管业务是否执行成功，都去记录执行日志（表：job_beforehand_retry），如果出现指定的异常，则标记记录为需要重试。待重试job执行时，就分发至相应的重试方法去执行。

**原理：**使用aop的方式，对需要重试的方法（含有@BeforehandRetry的方法）进行拦截

@BeforehandRetry:

/**
 * 事前补偿，确保每次执行业务时都有留底。会牺牲一性的性能。
 * @author wzy
 * @date 2017年7月14日 下午5:03:45
 */
@Target({ METHOD })
@Retention(RUNTIME)
@Inherited
public @interface BeforehandRetry {
	/**
	 * 执行重试的异常，默认是对BizRetryNeedException才去执行重试逻辑。业务异常是不需要重试的！！！
	 */
	Class<? extends Throwable> retryFor() default BizRetryNeedException.class;
	/**
	 * 最大重试次数
	 */
	int maxRetryCount() default 3;
}

**例子参考：**com.kvn.dal.core.beforehand_retry.BeforehandRetryBizService.java

@Service
public class BeforehandRetryBizService {
	@BeforehandRetry
	public String doBiz(Foo foo, String param){
		System.out.println("--->isRetryThread:" + ThreadContext.getContext().isRetryThread());
		System.out.println("参数：Foo=" + JSON.toJSONString(foo) + ", param=" + param);
		System.out.println("执行业务失败>>>>>>>>");
		throw new BizRetryNeedException("业务失败，需要重试！！！");
	}
}

事后重试

事后重试，即执行业务出现异常后，对于我们需要重试的异常，将重试参数持久化到DB（表：job_retry），然后通过事后重试调度定时发起重试。

对于需要重试的类，可以通过实现 IRetrySupport 接口，或者继承 AbstractRetrySupport 类。

IRetrySupport.java

public interface IRetrySupport {
	/**
	 * 重试
	 * @param retryContext 重试上下文
	 * @return 返回重试结果：true | false
	 */
	Boolean retry(AfterwardRetryContext retryContext);
}

例子参考：实现接口方式com.kvn.dal.core.afterward_retry.AfterwardRetryBizService.java

@Service
public class AfterwardRetryBizService implements IRetrySupport {
 @Resource
 IJobRetryDao jobRetryDao;
 public void executeBiz() {
 System.out.println(DateTime.now() + "--" + Thread.currentThread().getName() + "---------------doBizJob2222--------");
 try {
 Thread.sleep(3000L);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 Foo foo = new Foo(1001, "xxx");
 try {
 throw new RuntimeException("executeBiz异常，xxxxxxxxx");
 } catch (Exception e) {
 ArrayList<RetryParam> retryLs = new RetryParamListWrapper().buildRetryParam(foo).buildRetryParam("xxx").buildRetryParam("hehehe").toArrayList();
 JobRetry retry = JobRetry.createJobRetry(this.getClass(), "key001", retryLs);
 jobRetryDao.add(retry);
 throw e;
 }
 }
 @Override
 public Boolean retry(AfterwardRetryContext retryContext) {
 	/** 实现重试逻辑 */
 String retryDataKey = retryContext.getRetryDataKey();
 List<RetryParam> paramLs = retryContext.getRetryParamLs();
 Foo foo = paramLs.get(0).retoreParam(Foo.class);
 String originParam1 = paramLs.get(1).retoreParam(String.class);
 String originParam2 = paramLs.get(2).retoreParam(String.class);
 // 或者
 Foo foo2 = retryContext.getRetryParamValueMap().get(Foo.class).get(0);
 String originParam_1 = retryContext.getRetryParamValueMap().get(String.class).get(0);
 String originParam_2 = retryContext.getRetryParamValueMap().get(String.class).get(1);
 return true;
 }
}

例子参考：继承类的方式com.kvn.dal.core.afterward_retry.AfterwardRetryBestPracticeService.java

@Service
public class AfterwardRetryBestPracticeService extends AbstractRetrySupport {
	public void executeBiz() {
		System.out.println(DateTime.now() + "--" + Thread.currentThread().getName() + "---------------doBizJob2222--------");
		try {
			Thread.sleep(3000L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		Foo foo = new Foo(1001, "xxx");
		try {
			throw new RuntimeException("executeBiz异常，xxxxxxxxx");
		} catch (Exception e) {
			this.retryEnqueue("key001", foo, "hehe", "morning");
			throw e; // 出异常后，终止业务
		}
	}
	@Override
	public Boolean retry(AfterwardRetryContext retryContext) {
		// 实现重试逻辑
		return true;
	}
}