Java应用多机器部署解决大量定时任务问题

时间:2021-05-20

今天来说一个Java多机部署下定时任务的处理方案。

需求: 有两台服务器同时部署了同一套代码, 代码中写有spring自带的定时任务,但是每次执行定时任务时只需要一台机器去执行。

当拿到这个需求时我脑子中立马出现了两个简单的解决方案:

  • 利用ip进行判断, 两台机器ip肯定不一样, 指定某一台机器的ip运行。
  • 只在一台机器上部署定时任务的代码。

最后两个方案又都被自己否决了。 第一条,如果指定ip的机器出现了问题怎么办? 例如说宕机了, 那么该制定ip的机器上的定时任务是不是就无法运行了?如果以后该服务器迁移导致ip变化怎么办?

第二条, 同上, 还有就是要维护两套代码很不方便。

因为上面两个假设都不成立, 只能另找他法。 于是便想到利用mysql去解决, 之前了解过一点mysql的锁机制, 知道如果有同时的两个任务去写数据库中同一条记录, 只有一条会成功, 这是利用了mysql的排他锁。

下面就开始代码演示, 这里主要想给大家的是一个思路的提示, 代码还是很简单的。

首先需要单独创建一张表

CREATE TABLE `t_schedule_cluster` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '@cname:主键', `execute` int(1) NOT NULL COMMENT '@cname:执行状态', `version` int(11) NOT NULL COMMENT '@cname:版本号\r\n ', `task_name` varchar(128) NOT NULL COMMENT '@cname:任务名称\r\n ', `execute_ip` varchar(32) DEFAULT NULL COMMENT '@cname:执行ip\r\n ', `update_time` datetime DEFAULT NULL COMMENT '@cname:修改时间\r\n ', PRIMARY KEY (`id`), KEY `Index_series_id` (`execute`)) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8 COMMENT='@cname:多机定时任务调度';

看一下建成后的表结构, 注释写的已经很清楚了, 初始化时需要添加一些定时任务的名称(task_name), 这个要和你代码中保持一致, 后面会提到:

代码

首先看下我代码中用到的spring定时任务:

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://monConstants.ScheduleTaskName.LIST_CAR_TASK)) { //执行具体的task方法, doTask(); //将execute状态更新为0 scheduleClusterTask.end(LIST_CAR_TASK); } }

最后看下最核心的代码:ScheduleClusterTask.java

/** * 多机定时任务工具类 * Created by WangMeng on 2017/4/12. */@Servicepublic class ScheduleClusterTask { private ScheduleClusterEntityService scheduleClusterEntityService; /** * 这里因为两台机器都有同样的定时任务, 会同时执行这个方法,只有一台机器可以执行成功,返回true。 * @param maxExpireTime 最大的检查时间。 * @param taskName 任务名称。 * @return */ public boolean isValidMachine(long maxExpireTime, String taskName) { boolean isValid = false; try { //通过taskName去数据库中查找到该条记录, 如果大家使用的是mybatis这里需要改一下, 就是一个简单的查询操作 ScheduleClusterEntity carIndexEntity = scheduleClusterEntityService.findOne(ScheduleClusterEntity.Fields.taskName.eq(taskName)); int execute = carIndexEntity.getExecute(); String ip = InetAddress.getLocalHost().getHostAddress(); long currentTimeMillis = System.currentTimeMillis(); long time = carIndexEntity.getUpdateTime().getTime(); if (execute == 0 && time + maxExpireTime - 1000 < currentTimeMillis) { isValid = checkMachine(taskName, carIndexEntity, ip); } else if (time + maxExpireTime - 1000 < currentTimeMillis){ //这里要判断下, 如果上一次执行出现异常导致execute没有更新为0, 那么这里要判断上一次更新时间的间隔。 isValid = checkMachine(taskName, carIndexEntity, ip); } } catch (UnknownHostException e) { e.printStackTrace(); } return isValid; } /** * end方法主要是将excute(是否正在执行的标志位,0:没有执行, 1:正在执行)更新为0 * @param taskName * @return */ public boolean end (String taskName) { ScheduleClusterEntity carIndexEntity = scheduleClusterEntityService.findOne(ScheduleClusterEntity.Fields.taskName.eq(taskName)); //将execute状态更新为0 return scheduleClusterEntityService.end(carIndexEntity); } private boolean checkMachine(String taskName, ScheduleClusterEntity carIndexRefresh, String ip) { return scheduleClusterEntityService.start(taskName, carIndexRefresh.getVersion(), ip); } @Autowired public void setScheduleClusterEntityService(ScheduleClusterEntityService scheduleClusterEntityService) { this.scheduleClusterEntityService = scheduleClusterEntityService; }}

这里还有start方法, 看看怎样的操作:

@Repositorypublic class DefaultScheduleClusterEntityDao extends AbstractDao<ScheduleClusterEntity> implements ScheduleClusterEntityDao { @Override public boolean start(String taskName, int version, String ip) { String sql = "update t_schedule_cluster set execute = 1, " + "version = ?, execute_ip = ?, update_time = ?" + " where task_name = ? and version = ?"; Sql s = new Sql(sql); s.addParam(version + 1); s.addParam(ip); s.addParam(SqlTimeUtils.nowTimestamp()); s.addParam(taskName); s.addParam(version); return 1 == executeUpdate(s); }}

核心的代码到了这里就没有了, 代码确实是非常非常的简单, 有兴趣的话大家可以在本地测试一下就可以。

当然还有更多很好地解决方案, 我这里秉承的是最简单的处理方式。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。

声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。

相关文章