Geplante Spring-Task, die in einer Clusterumgebung ausgeführt wird

Ich schreibe eine Anwendung mit einem Cron-Job, der alle 60 Sekunden ausgeführt wird. Die Anwendung ist so konfiguriert, dass sie bei Bedarf auf mehrere Instanzen skaliert werden kann. Ich möchte die Aufgabe nur auf einer Instanz alle 60 Sekunden ausführen (auf einem beliebigen Knoten). Ich kann keine Lösung für dieses Problem finden und bin überrascht, dass es zuvor nicht mehrmals gefragt wurde. Ich benutze Spring 4.1.6.

    <task:scheduled-tasks>
        <task:scheduled ref="beanName" method="execute" cron="0/60 * * * * *"/>
    </task:scheduled-tasks>

Es gibt ein ShedLock- Projekt, das genau diesem Zweck dient. Sie kommentieren nur Aufgaben, die bei der Ausführung gesperrt werden sollen

@Scheduled( ... )
@SchedulerLock(name = "scheduledTaskName")
public void scheduledTask() {
   // do something
}

Konfigurieren Sie Spring und einen LockProvider

@Configuration
@EnableScheduling
@EnableSchedulerLock(defaultLockAtMostFor = "10m")
class MySpringConfiguration {
    ...
    @Bean
    public LockProvider lockProvider(DataSource dataSource) {
       return new JdbcTemplateLockProvider(dataSource);
    }
    ...
}

Ich denke, Sie müssen Quartz Clustering mit JDBC-JobStore für diesen Zweck verwenden

Dies ist eine weitere einfache und robuste Methode, um einen Job in einem Cluster sicher auszuführen. Sie können basierend auf der Datenbank die Aufgabe nur ausführen, wenn der Knoten der "Leiter" im Cluster ist.

Auch wenn ein Knoten ausfällt oder im Cluster heruntergefahren wird, wird ein anderer Knoten zum Leader.

Alles, was Sie haben, ist, einen "Führerwahl" -Mechanismus zu erstellen und jedes Mal zu überprüfen, ob Sie der Führer sind:

@Scheduled(cron = "*/30 * * * * *")
public void executeFailedEmailTasks() {
    if (checkIfLeader()) {
        final List<EmailTask> list = emailTaskService.getFailedEmailTasks();
        for (EmailTask emailTask : list) {
            dispatchService.sendEmail(emailTask);
        }
    }
}

Befolgen Sie diese Schritte:

1. Definieren Sie das Objekt und die Tabelle, die einen Eintrag pro Knoten im Cluster enthalten:

@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.Erstellen Sie den Dienst, der a) den Knoten in die Datenbank einfügt, b) nach Leader sucht

@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 der Datenbank, um zu senden, dass Sie am Leben sind

@Override
@Scheduled(cron = "0 0/5 * * * ?")
public void executeSystemNodePing() {
    systemNodeService.pingNode();
}

@Override
@Scheduled(cron = "0 0/10 * * * ?")
public void executeLeaderResolution() {
    systemNodeService.checkLeaderShip();
}

4.Sie sind bereit! Überprüfen Sie einfach, ob Sie der Anführer sind, bevor Sie die Aufgabe ausführen:

@Override
@Scheduled(cron = "*/30 * * * * *")
public void executeFailedEmailTasks() {
    if (checkIfLeader()) {
        final List<EmailTask> list = emailTaskService.getFailedEmailTasks();
        for (EmailTask emailTask : list) {
            dispatchService.sendEmail(emailTask);
        }
    }
}

Stapel- und geplante Jobs werden normalerweise auf ihren eigenen Standalone-Servern ausgeführt, weg von kundenorientierten Apps. Daher ist es nicht üblich, einen Job in eine Anwendung aufzunehmen, die voraussichtlich in einem Cluster ausgeführt wird. Darüber hinaus müssen sich Jobs in Clusterumgebungen normalerweise nicht um andere Instanzen desselben Jobs kümmern, die parallel ausgeführt werden. Dies ist ein weiterer Grund, warum die Isolierung von Jobinstanzen keine große Anforderung ist.

Eine einfache Lösung wäre, Ihre Jobs in einem Spring-Profil zu konfigurieren. Zum Beispiel, wenn Ihre aktuelle Konfiguration lautet:

<beans>
  <bean id="someBean" .../>

  <task:scheduled-tasks>
    <task:scheduled ref="someBean" method="execute" cron="0/60 * * * * *"/>
  </task:scheduled-tasks>
</beans>

ändere es in:

<beans>
  <beans profile="scheduled">
    <bean id="someBean" .../>

    <task:scheduled-tasks>
      <task:scheduled ref="someBean" method="execute" cron="0/60 * * * * *"/>
    </task:scheduled-tasks>
  </beans>
</beans>

Starten Sie dann Ihre Anwendung auf nur einem Computer mit scheduledaktiviertem Profil ( -Dspring.profiles.active=scheduled).

Wenn der Primärserver aus irgendeinem Grund nicht mehr verfügbar ist, starten Sie einfach einen anderen Server mit aktiviertem Profil, und die Dinge funktionieren weiterhin einwandfrei.

Die Dinge ändern sich, wenn Sie auch für die Jobs ein automatisches Failover wünschen. Anschließend müssen Sie den Job auf allen Servern ausführen und die Synchronisierung über eine gemeinsame Ressource wie eine Datenbanktabelle, einen Cluster-Cache, eine JMX-Variable usw. überprüfen.

dlock ist so konzipiert, dass Aufgaben nur einmal ausgeführt werden, indem Datenbankindizes und -einschränkungen verwendet werden. Sie können einfach so etwas wie unten tun.

@Scheduled(cron = "30 30 3 * * *")
@TryLock(name = "executeMyTask", owner = SERVER_NAME, lockFor = THREE_MINUTES)
public void execute() {

}

Lesen Sie den Artikel über die Verwendung.

Ich verwende eine Datenbanktabelle, um das Sperren durchzuführen. Es kann jeweils nur eine Aufgabe in die Tabelle eingefügt werden. Der andere erhält eine DuplicateKeyException. Die Einfüge- und Löschlogik wird durch einen Aspekt um die Annotation @Scheduled behandelt. Ich benutze 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)
);

Sie können dazu einen einbettbaren Scheduler wie db-scheduler verwenden . Es verfügt über dauerhafte Ausführungen und verwendet einen einfachen optimistischen Sperrmechanismus, um die Ausführung durch einen einzelnen Knoten zu gewährleisten.

Beispielcode, wie der Anwendungsfall erreicht werden kann:

   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();

Der Spring-Kontext ist nicht geclustert, daher ist die Verwaltung der Aufgabe in verteilten Anwendungen etwas schwierig. Sie müssen Systeme verwenden, die jgroup unterstützen, um den Status zu synchronisieren und Ihre Aufgabe die Priorität zum Ausführen der Aktion übernehmen zu lassen. Oder könnten Sie ejb Kontext verwenden Cluster - Verwaltung ha Singleton Service wie Jboss ha Umgebung https://developers.redhat.com/quickstarts/eap/cluster-ha-singleton/?referrer=jbd Oder könnten Sie gruppierten Cache und Zugriffssperre Ressource verwenden Zwischen dem Dienst und dem ersten Dienst wird die Sperre die Aktion bilden oder eine eigene jgroup implementieren, um Ihren Dienst zu kommunizieren und die Aktion an einem Knoten auszuführen