我正在编写一个具有每 60 秒执行一次的 cron 作业的应用程序。该应用程序配置为在需要时扩展到多个实例。我只想每 60 秒在 1 个实例上执行任务（在任何节点上）。开箱即用我找不到解决方案，我很惊讶以前没有被问过多次。我正在使用 Spring 4.1.6。
<task:scheduled-tasks>
<task:scheduled ref="beanName" method="execute" cron="0/60 * * * * *"/>
</task:scheduled-tasks>
有一个ShedLock项目正是为这个目的服务的。您只需注释执行时应锁定的任务
@Scheduled( ... )
@SchedulerLock(name = "scheduledTaskName")
public void scheduledTask() {
// do something
}
配置 Spring 和 LockProvider
@Configuration
@EnableScheduling
@EnableSchedulerLock(defaultLockAtMostFor = "10m")
class MySpringConfiguration {
...
@Bean
public LockProvider lockProvider(DataSource dataSource) {
return new JdbcTemplateLockProvider(dataSource);
}
...
}
我认为您必须为此目的使用Quartz Clustering 和 JDBC-JobStore
这是在集群中安全执行作业的另一种简单而强大的方法。只有当节点是集群中的“领导者”时，您才能基于数据库执行任务。
此外，当集群中的一个节点发生故障或关闭时，另一个节点成为领导者。
您所要做的就是创建一个“领导者选举”机制，并且每次都检查您是否是领导者：
@Scheduled(cron = "*/30 * * * * *")
public void executeFailedEmailTasks() {
if (checkIfLeader()) {
final List<EmailTask> list = emailTaskService.getFailedEmailTasks();
for (EmailTask emailTask : list) {
dispatchService.sendEmail(emailTask);
}
}
}
请按照以下步骤操作：
1.定义在集群中每个节点保存一个条目的对象和表：
@Entity(name = "SYS_NODE")
public class SystemNode {
/** The id. */
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
/** The name. */
@Column(name = "TIMESTAMP")
private String timestamp;
/** The ip. */
@Column(name = "IP")
private String ip;
/** The last ping. */
@Column(name = "LAST_PING")
private Date lastPing;
/** The last ping. */
@Column(name = "CREATED_AT")
private Date createdAt = new Date();
/** The last ping. */
@Column(name = "IS_LEADER")
private Boolean isLeader = Boolean.FALSE;
public Long getId() {
return id;
}
public void setId(final Long id) {
this.id = id;
}
public String getTimestamp() {
return timestamp;
}
public void setTimestamp(final String timestamp) {
this.timestamp = timestamp;
}
public String getIp() {
return ip;
}
public void setIp(final String ip) {
this.ip = ip;
}
public Date getLastPing() {
return lastPing;
}
public void setLastPing(final Date lastPing) {
this.lastPing = lastPing;
}
public Date getCreatedAt() {
return createdAt;
}
public void setCreatedAt(final Date createdAt) {
this.createdAt = createdAt;
}
public Boolean getIsLeader() {
return isLeader;
}
public void setIsLeader(final Boolean isLeader) {
this.isLeader = isLeader;
}
@Override
public String toString() {
return "SystemNode{" +
"id=" + id +
", timestamp='" + timestamp + '\'' +
", ip='" + ip + '\'' +
", lastPing=" + lastPing +
", createdAt=" + createdAt +
", isLeader=" + isLeader +
'}';
}
}
2.创建服务a）将节点插入数据库，b）检查领导者
@Service
@Transactional
public class SystemNodeServiceImpl implements SystemNodeService, ApplicationListener {
/** The logger. */
private static final Logger LOGGER = Logger.getLogger(SystemNodeService.class);
/** The constant NO_ALIVE_NODES. */
private static final String NO_ALIVE_NODES = "Not alive nodes found in list {0}";
/** The ip. */
private String ip;
/** The system service. */
private SystemService systemService;
/** The system node repository. */
private SystemNodeRepository systemNodeRepository;
@Autowired
public void setSystemService(final SystemService systemService) {
this.systemService = systemService;
}
@Autowired
public void setSystemNodeRepository(final SystemNodeRepository systemNodeRepository) {
this.systemNodeRepository = systemNodeRepository;
}
@Override
public void pingNode() {
final SystemNode node = systemNodeRepository.findByIp(ip);
if (node == null) {
createNode();
} else {
updateNode(node);
}
}
@Override
public void checkLeaderShip() {
final List<SystemNode> allList = systemNodeRepository.findAll();
final List<SystemNode> aliveList = filterAliveNodes(allList);
SystemNode leader = findLeader(allList);
if (leader != null && aliveList.contains(leader)) {
setLeaderFlag(allList, Boolean.FALSE);
leader.setIsLeader(Boolean.TRUE);
systemNodeRepository.save(allList);
} else {
final SystemNode node = findMinNode(aliveList);
setLeaderFlag(allList, Boolean.FALSE);
node.setIsLeader(Boolean.TRUE);
systemNodeRepository.save(allList);
}
}
/**
* Returns the leaded
* @param list
* the list
* @return the leader
*/
private SystemNode findLeader(final List<SystemNode> list) {
for (SystemNode systemNode : list) {
if (systemNode.getIsLeader()) {
return systemNode;
}
}
return null;
}
@Override
public boolean isLeader() {
final SystemNode node = systemNodeRepository.findByIp(ip);
return node != null && node.getIsLeader();
}
@Override
public void onApplicationEvent(final ApplicationEvent applicationEvent) {
try {
ip = InetAddress.getLocalHost().getHostAddress();
} catch (Exception e) {
throw new RuntimeException(e);
}
if (applicationEvent instanceof ContextRefreshedEvent) {
pingNode();
}
}
/**
* Creates the node
*/
private void createNode() {
final SystemNode node = new SystemNode();
node.setIp(ip);
node.setTimestamp(String.valueOf(System.currentTimeMillis()));
node.setCreatedAt(new Date());
node.setLastPing(new Date());
node.setIsLeader(CollectionUtils.isEmpty(systemNodeRepository.findAll()));
systemNodeRepository.save(node);
}
/**
* Updates the node
*/
private void updateNode(final SystemNode node) {
node.setLastPing(new Date());
systemNodeRepository.save(node);
}
/**
* Returns the alive nodes.
*
* @param list
* the list
* @return the alive nodes
*/
private List<SystemNode> filterAliveNodes(final List<SystemNode> list) {
int timeout = systemService.getSetting(SettingEnum.SYSTEM_CONFIGURATION_SYSTEM_NODE_ALIVE_TIMEOUT, Integer.class);
final List<SystemNode> finalList = new LinkedList<>();
for (SystemNode systemNode : list) {
if (!DateUtils.hasExpired(systemNode.getLastPing(), timeout)) {
finalList.add(systemNode);
}
}
if (CollectionUtils.isEmpty(finalList)) {
LOGGER.warn(MessageFormat.format(NO_ALIVE_NODES, list));
throw new RuntimeException(MessageFormat.format(NO_ALIVE_NODES, list));
}
return finalList;
}
/**
* Finds the min name node.
*
* @param list
* the list
* @return the min node
*/
private SystemNode findMinNode(final List<SystemNode> list) {
SystemNode min = list.get(0);
for (SystemNode systemNode : list) {
if (systemNode.getTimestamp().compareTo(min.getTimestamp()) < -1) {
min = systemNode;
}
}
return min;
}
/**
* Sets the leader flag.
*
* @param list
* the list
* @param value
* the value
*/
private void setLeaderFlag(final List<SystemNode> list, final Boolean value) {
for (SystemNode systemNode : list) {
systemNode.setIsLeader(value);
}
}
}
3.ping数据库发送你还活着
@Override
@Scheduled(cron = "0 0/5 * * * ?")
public void executeSystemNodePing() {
systemNodeService.pingNode();
}
@Override
@Scheduled(cron = "0 0/10 * * * ?")
public void executeLeaderResolution() {
systemNodeService.checkLeaderShip();
}
4.你准备好了！只需在执行任务之前检查您是否是领导者：
@Override
@Scheduled(cron = "*/30 * * * * *")
public void executeFailedEmailTasks() {
if (checkIfLeader()) {
final List<EmailTask> list = emailTaskService.getFailedEmailTasks();
for (EmailTask emailTask : list) {
dispatchService.sendEmail(emailTask);
}
}
}
批处理和计划作业通常在它们自己的独立服务器上运行，远离面向客户的应用程序，因此将作业包含在预期在集群上运行的应用程序中并不是一个常见的要求。此外，集群环境中的作业通常不需要担心同一作业的其他实例并行运行，因此作业实例的隔离不是一个大要求的另一个原因。
一个简单的解决方案是在 Spring Profile 中配置您的作业。例如，如果您当前的配置是：
<beans>
<bean id="someBean" .../>
<task:scheduled-tasks>
<task:scheduled ref="someBean" method="execute" cron="0/60 * * * * *"/>
</task:scheduled-tasks>
</beans>
将其更改为：
<beans>
<beans profile="scheduled">
<bean id="someBean" .../>
<task:scheduled-tasks>
<task:scheduled ref="someBean" method="execute" cron="0/60 * * * * *"/>
</task:scheduled-tasks>
</beans>
</beans>
然后，在激活配置文件的一台机器上启动您的应用程序scheduled( -Dspring.profiles.active=scheduled)。
如果主服务器由于某种原因变得不可用，只需启动另一台启用配置文件的服务器，一切都会继续正常工作。
如果您还想对作业进行自动故障转移，情况就会发生变化。然后，您需要保持作业在所有服务器上运行，并通过数据库表、集群缓存、JMX 变量等公共资源检查同步。
我正在使用数据库表进行锁定。一次只有一项任务可以插入到表中。另一个将获得 DuplicateKeyException。插入和删除逻辑由@Scheduled 注释周围的方面处理。我正在使用 Spring Boot 2.0
@Component
@Aspect
public class SchedulerLock {
private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerLock.class);
@Autowired
private JdbcTemplate jdbcTemplate;
@Around("execution(@org.springframework.scheduling.annotation.Scheduled * *(..))")
public Object lockTask(ProceedingJoinPoint joinPoint) throws Throwable {
String jobSignature = joinPoint.getSignature().toString();
try {
jdbcTemplate.update("INSERT INTO scheduler_lock (signature, date) VALUES (?, ?)", new Object[] {jobSignature, new Date()});
Object proceed = joinPoint.proceed();
jdbcTemplate.update("DELETE FROM scheduler_lock WHERE lock_signature = ?", new Object[] {jobSignature});
return proceed;
}catch (DuplicateKeyException e) {
LOGGER.warn("Job is currently locked: "+jobSignature);
return null;
}
}
}
@Component
public class EveryTenSecondJob {
@Scheduled(cron = "0/10 * * * * *")
public void taskExecution() {
System.out.println("Hello World");
}
}
CREATE TABLE scheduler_lock(
signature varchar(255) NOT NULL,
date datetime DEFAULT NULL,
PRIMARY KEY(signature)
);
您可以在这里使用 Zookeeper 来选举主实例，主实例将只运行计划的作业。
这里的一种实现是使用 Aspect 和 Apache Curator
@SpringBootApplication
@EnableScheduling
public class Application {
private static final int PORT = 2181;
@Bean
public CuratorFramework curatorFramework() {
CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:" + PORT, new ExponentialBackoffRetry(1000, 3));
client.start();
return client;
}
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
方面类
@Aspect
@Component
public class LeaderAspect implements LeaderLatchListener{
private static final Logger log = LoggerFactory.getLogger(LeaderAspect.class);
private static final String ELECTION_ROOT = "/election";
private volatile boolean isLeader = false;
@Autowired
public LeaderAspect(CuratorFramework client) throws Exception {
LeaderLatch ll = new LeaderLatch(client , ELECTION_ROOT);
ll.addListener(this);
ll.start();
}
@Override
public void isLeader() {
isLeader = true;
log.info("Leadership granted.");
}
@Override
public void notLeader() {
isLeader = false;
log.info("Leadership revoked.");
}
@Around("@annotation(com.example.apache.curator.annotation.LeaderOnly)")
public void onlyExecuteForLeader(ProceedingJoinPoint joinPoint) {
if (!isLeader) {
log.debug("I'm not leader, skip leader-only tasks.");
return;
}
try {
log.debug("I'm leader, execute leader-only tasks.");
joinPoint.proceed();
} catch (Throwable ex) {
log.error(ex.getMessage());
}
}
}
LeaderOnly注解
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface LeaderOnly {
}
计划任务
@Component
public class HelloWorld {
private static final Logger log = LoggerFactory.getLogger(HelloWorld.class);
@LeaderOnly
@Scheduled(fixedRate = 1000L)
public void sayHello() {
log.info("Hello, world!");
}
}
您可以使用db-scheduler 之类的可嵌入调度程序来完成此任务。它具有持久执行并使用简单的乐观锁定机制来保证由单个节点执行。
如何实现用例的示例代码：
RecurringTask<Void> recurring1 = Tasks.recurring("my-task-name", FixedDelay.of(Duration.ofSeconds(60)))
.execute((taskInstance, executionContext) -> {
System.out.println("Executing " + taskInstance.getTaskAndInstance());
});
final Scheduler scheduler = Scheduler
.create(dataSource)
.startTasks(recurring1)
.build();
scheduler.start();
Spring上下文不是集群的，所以在分布式应用中管理任务有点困难，你需要使用支持jgroup的系统来同步状态，让你的任务优先执行动作。或者您可以使用 ejb 上下文来管理集群 ha 单例服务，例如 jboss ha 环境 https://developers.redhat.com/quickstarts/eap/cluster-ha-singleton/?referrer=jbd 或者您可以使用集群缓存和访问锁资源在服务和第一个服务之间获取锁定将形成操作或实现您自己的 jgroup 来与您的服务通信并执行操作一个节点