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

92

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>
user3131879
quelle
7
Ich denke, Quarz ist die beste Lösung für Sie: stackoverflow.com/questions/6663182/…
Selalerer
Irgendwelche Vorschläge zur Verwendung CronJobin kubernetes?
ch271828n

Antworten:

94

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);
    }
    ...
}
Lukas
quelle
1
Ich möchte nur "Gute Arbeit!" Sagen. Aber ... Die nette Funktion ist, wenn die Bibliothek den Datenbanknamen ermitteln kann, ohne ihn explizit im Code anzugeben ... Außer, dass er hervorragend funktioniert!
Krzysiek
Funktioniert für mich mit Oracle und Spring Boot Data JPA Starter.
Mahendran Ayyarsamy Kandiar
Funktioniert diese Lösung für Spring 3.1.1.RELEASE und Java 6? Bitte erzähle.
Vikas Sharma
Ich habe es mit MsSQL und Spring Boot JPA versucht und ich habe ein Liquibase-Skript für den SQL-Teil verwendet. Funktioniert gut. Danke
Blatt
Es funktioniert in der Tat gut. Allerdings habe ich hier einen etwas komplexen Fall getroffen. Könnten Sie bitte einen Blick darauf werfen? Vielen Dank!!! stackoverflow.com/questions/57691205/…
Dayton Wang
15

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);
        }
    }
}
mspapant
quelle
Was ist in diesem Fall SystemService und SettingEnum? Sieht so aus, als wäre es extrem einfach und gibt nur einen Timeout-Wert zurück. In diesem Fall, warum nicht einfach das Timeout hart codieren?
Tlavarea
@mspapant, was ist das SettingEnum.SYSTEM_CONFIGURATION_SYSTEM_NODE_ALIVE_TIMEOUT? Was ist der optimale Wert, den ich hier verwenden sollte?
user525146
@tlavarea Haben Sie diesen Code implementiert? Ich habe eine Frage zur DateUtils.hasExpired-Methode. Ist es eine benutzerdefinierte Methode oder ist es ein Apache Common Utils?
user525146
11

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.

manisch
quelle
55
Dies ist eine gültige Problemumgehung, die jedoch gegen die Idee einer Clusterumgebung verstößt. Wenn ein Knoten ausfällt, kann der andere Knoten andere Anforderungen bedienen. In dieser Problemumgehung wird dieser Hintergrundjob nicht ausgeführt, wenn der Knoten mit dem "geplanten" Profil ausfällt
Ahmed Hashem
3
Ich denke, wir könnten Redis mit Atomic getund setOperation verwenden, um dies zu erreichen.
Thanh Nguyen Van
Es gibt mehrere Probleme mit Ihrem Vorschlag: 1. Sie möchten im Allgemeinen, dass jeder Knoten eines Clusters genau dieselbe Konfiguration hat, sodass er zu 100% austauschbar ist und dieselben Ressourcen unter derselben Last benötigt, die er gemeinsam nutzt. 2. Ihre Lösung würde einen manuellen Eingriff erfordern, wenn der "Task" -Knoten ausfällt. 3. Es kann immer noch nicht garantiert werden, dass der Job tatsächlich erfolgreich ausgeführt wurde, da der Knoten "Task" ausgefallen ist, bevor die Verarbeitung der aktuellen Ausführung abgeschlossen ist, und der neue "Task Runner" erstellt wurde, nachdem der erste ausgefallen ist, ohne zu wissen, ob es war fertig oder nicht.
Moshe Bixenshpaner
1
Es verstößt einfach gegen die Idee von Clusterumgebungen. Mit dem von Ihnen vorgeschlagenen Ansatz kann es keine Lösung geben. Sie können nicht einmal die Profilserver replizieren, um die Verfügbarkeit sicherzustellen, da dies zu zusätzlichen Kosten und unnötiger Verschwendung von Ressourcen führt. Die von @Thanh vorgeschlagene Lösung ist viel sauberer. Stellen Sie sich das gleiche wie einen MUTEX vor. Jeder Server, auf dem das Skript ausgeführt wird, erhält eine temporäre Sperre in einem verteilten Cache wie Redis und fährt dann mit den Konzepten der herkömmlichen Sperre fort.
Anuj Pradhan
2

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.

Will Hughes
quelle
3
Wenn Sie dlock verwenden. Nehmen wir an, wir verwenden DB, um die Sperre aufrechtzuerhalten. Und einer der Knoten im Cluster ist nach dem Sperren unerwartet ausgefallen. Was passiert dann in diesem Szenario? Wird es in einem Deadlock-Zustand sein?
Badman
1

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)
);
RenRen
quelle
3
Glaubst du, es wird perfekt funktionieren? Denn wenn einer der Knoten nach dem Sperren ausfällt, erfahren andere nicht, warum es eine Sperre gibt (in Ihrem Fall entspricht der Zeileneintrag dem Job in der Tabelle).
Badman
0

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();
Gustav Karlsson
quelle
-1

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

Abdulghaffar Al-Labadi
quelle