Zamanlanmış Görevlerin Quartz ile Dinamik Yönetimi — 2. Bölüm

İki bölümden oluşan yazımızın ilk bölümünde, Quartz kütüphanesinin temel bileşenlerini öğrendik ve örnek bir Spring uygulamasında belirli zamanlarda çalışmasını istediğimiz iş parçacıklarını nasıl çalıştıracağımızı gördük.

İlk bölüme aşağıdaki linkten ulaşabilirsiniz:

https://medium.com/akbank-teknoloji/zamanlanm%C4%B1%C5%9F-g%C3%B6revlerin-quartz-ile-dinamik-y%C3%B6netimi-1-b%C3%B6l%C3%BCm-2dfc09aabc87

İkinci bölümümüzde ise ilk bölümde öğrendiklerimizi pratiğe dökeceğimiz mini bir ihale sistemi yazacağız.

Haydi, başlayalım!

Örnek proje (Mini İhale Sistemi)

Örnek proje olarak Spring Boot kullanarak mini bir ihale sistemi yazacağız. Belirli periyotlarda veri tabanında bulunan ihaleleri ve bu ihalelere verilen teklifleri tarayarak en yüksek teklifi bulacağız. Söz konusu işi belirli periyotlarda veya istenilen zamanlarda yapabilmek için Quartz’ı devreye sokacağız.

Projeyi aşağıdaki konfigürasyonda oluşturabilmek için https://start.spring.io/ sitesini kullanacağız.

Öncelikle, Quartz kütüphanesini autoconfigured bir şekilde uygulamamızda kullanabilmek için projeye ekliyoruz. Ayrıca, projede Job’ların dinamik olarak yönetilmesi için bir rest api yazacağımızdan Spring Web, Java sınıflarında getter, setter vb. basmakalıp kodları elle yazmamak için Lombok, veri tabanı olarak PostgreSQL kullanacağımız için PostgreSQL Driver, persistent çatısı olarak da Hibernate kullanacağımız için Spring Data JPA bağımlılıklarını projeye ekliyoruz.

pom.xml bağımlılıklarımız aşağıdaki gibi olmalı.

<dependencies>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-jpa</artifactId>
  </dependency>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-quartz</artifactId>
  </dependency>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency>
   <groupId>org.postgresql</groupId>
   <artifactId>postgresql</artifactId>
   <scope>runtime</scope>
  </dependency>
  <dependency>
   <groupId>org.projectlombok</groupId>
   <artifactId>lombok</artifactId>
   <optional>true</optional>
  </dependency>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
  </dependency>
 </dependencies>

Uygulamada ihaleleri temsil eden java sınıfını yazıyoruz.

// Auction.java
package com.kocesat.quartzdemo.model;

import jakarta.persistence.*;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.io.Serializable;
import java.math.BigDecimal;
import java.time.LocalDateTime;

@Entity
@Table(name = "auctions")
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class Auction implements Serializable {
  private static final long serialVersionUID = 1L;
  @Id
  @GeneratedValue(strategy = GenerationType.IDENTITY)
  private Integer id;
  private String title;
  private LocalDateTime closingTime;
  private Integer status; // 1-açık, 0-kapalı
  private String bids; // "23.50|14.30|200.00" şeklinde tutacağız
  private BigDecimal winnerPrice;
}

Bir ihaleye verilen teklifleri String tipinde (“23.50|14.30|200.00”) | ayracını kullanarak tutacağız. Ancak, ciddi bir uygulamada bu şekilde tutulması tercih edilmez, tekliflerin ayrı birer tablosu olmalıdır. (Biz bu karmaşıklığa şu an için girmeyeceğiz.)

İhale sınıfının veri tabanı işlemlerini gerçekleştirecek repository sınıfını yazıyoruz.

// AuctionRepository.java
package com.kocesat.quartzdemo.repository;

import com.kocesat.quartzdemo.model.Auction;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

import java.util.List;

@Repository
public interface AuctionRepository 
  extends JpaRepository<Auction, Integer> {

  List<Auction> findAuctionByStatus(Integer status);
}

Açık ihale kayıtlarını getirecek, kazanan teklifi belirleyerek ihaleyi kapatan servis sınıfını AuctionService yazıyoruz.

// AuctionService.java
package com.kocesat.quartzdemo.service;

import com.kocesat.quartzdemo.exception.AppRuntimeException;
import com.kocesat.quartzdemo.model.Auction;
import com.kocesat.quartzdemo.repository.AuctionRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;

@Service
@Transactional(rollbackFor = Throwable.class)
@RequiredArgsConstructor
public class AuctionService {
  private final AuctionRepository repository;

  @Transactional(readOnly = true)
  public List<Auction> getOpenAuctions() {
    return repository.findAuctionByStatus(1);
  }

  @Transactional(propagation = Propagation.REQUIRES_NEW)
  public void decideWinner(Integer id) {
    final Auction auction = repository
      .findById(id)
      .orElseThrow(AppRuntimeException::new);
    auction.setWinnerPrice(winnerPrice(auction.getBids()));
    auction.setStatus(0);
    repository.save(auction);
  }

  private BigDecimal winnerPrice(String bids) {
    if (bids == null) {
      return BigDecimal.ZERO;
    }
    return Arrays.stream(bids.split("\\|"))
      .map(BigDecimal::new)
      .max(BigDecimal::compareTo)
      .orElseThrow(AppRuntimeException::new);
  }

}

İhaleleri tutacak veritabanını oluşturup, ilgili ihale kayıtlarını insert ediyoruz.

CREATE TABLE auctions (
  id serial NOT NULL,
  title varchar(255),
  closing_time timestamp(6),
  status integer,
  bids varchar(255),
  winner_price numeric(38,2),
  PRIMARY KEY (id)
)

INSERT INTO auctions (title, closing_time, status, bids)
VALUES
    ('Tesla İhalesi', '2022/02/27 23:00', 1, '2300.50|1400.30|2000.00'),
    ('Macbook İhalesi', '2022/02/27 17:00', 1, '100.50|120.30|90.00'),
    ('Iphone İhalesi', '2022/02/27 23:00', 1, '34.50|82.30|62.00')

İhale bitiş zamanı (closingTime) alanını demoyu denediğiniz zamana göre değiştirebilirsiniz.

Aşağıdaki sql’i çalıştırdığımızda kayıtların geldiğini görmeliyiz.

select * from auctions;

Uygulamamızı veri tabanına bağlayabilmek için application.properties dosyasına aşağıdaki konfigürasyonları ekliyoruz.

#application.properties
spring.datasource.url=jdbc:postgresql://localhost:5432/postgres
spring.datasource.username=${USERNAME}
spring.datasource.password=${PASSWORD}
spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.PostgreSQLDialect
spring.jpa.hibernate.ddl-auto=update
spring.jpa.properties.hibernate.show_sql=true
spring.jpa.properties.hibernate.use_sql_comments=true
spring.jpa.properties.hibernate.format_sql=true

Daha önce bahsettiğimiz üzere, Quartz Scheduler bilgileri bellekte tutabildiği gibi veri tabanında da tutabiliyordu. Bu özelliği veri tabanı olarak ayarlamak ve diğer Quartz konfigürasyonlarını yapmak için application.properties dosyasına aşağıdaki ayarları ekliyoruz.

## Quartz Properties
# To persist the jobs in the database
spring.quartz.job-store-type=jdbc

# The number of concurrent jobs running at the same time
# defaults to 10
spring.quartz.properties.org.quartz.threadPool.threadCount=5

# the name of scheduler instance for this application
spring.quartz.scheduler-name=demoapp

# Tell Quartz for this app to run in the clustered environment.
spring.quartz.properties.org.quartz.jobStore.isClustered=true
spring.quartz.properties.org.quartz.scheduler.instanceId=AUTO

# To make Spring Boot to generate Quartz tables
spring.quartz.jdbc.initialize-schema=always

spring.quartz.properties.org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.PostgreSQLDelegate

# To give prefixes to quartz tables. Defaults to QRTZ_
spring.quartz.properties.org.quartz.jobStore.tablePrefix=QRTZ_

# The number of milliseconds the scheduler will tolerate a trigger to pass its next-fire-time by,
# before being considered misfired
# The default value (if you don?t make an entry of this property in your configuration) is 60000 (60 seconds).
spring.quartz.properties.org.quartz.jobStore.misfireThreshold=5000

# To make JobDataMap to use String key values
spring.quartz.properties.org.quartz.jobStore.useProperties=true

Store tipi olarak veri tabanı kullanacağımız için postgre’de aşağıdaki Quartz tablolarını eklememiz gerekiyor.

Quartz tabloları

`qrtz` prefixi ile başlayan tablolar tamamen Quartz tarafından kullanılacak olup, uygulama içinden veya dışından (Zorunlu olmadıkça) müdahale edilmemesini öneriyoruz.

Varsayılan olarak ‘qrtz’ olan tablo prefixini değiştirmek isterseniz, aşağıdaki konfigürasyonu da ekleyebilirsiniz. (Tabii bu durumda tabloları oluştururken uygun prefix ile oluşturmanız gerekecektir.)

# application.properties
org.quartz.jobStore.tablePrefix=QRTZ_

İlk görevimizi yazmak için tüm hazırlıklarımız tamam. Şimdi, daha sonra belirleyeceğimiz bir saatte çalışmasını isteyeceğimiz görevin yapacağı işi, yani DetermineAuctionWinnerJob sınıfımızı yazalım.

// DetermineAuctionWinnerJob.java
package com.kocesat.quartzdemo.job;

import com.kocesat.quartzdemo.model.Auction;
import com.kocesat.quartzdemo.service.AuctionService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.util.List;

@Component
@DisallowConcurrentExecution
@RequiredArgsConstructor
@Slf4j
public class DetermineAuctionWinnerJob extends QuartzJobBean {
  private final AuctionService auctionService;
  @Override
  protected void executeInternal(JobExecutionContext context) 
    throws JobExecutionException {
    log.info("DetermineAuctionWinnerJob fired at " + LocalDateTime.now());
    try {
      final List<Auction> auctions = auctionService.getOpenAuctions();
      if (auctions.isEmpty()) {
        log.info("No open auctions found. Returning!!!");
      }
      auctions.forEach(auction -> {
        log.info(auction.toString());
        if (auction.getClosingTime().isBefore(LocalDateTime.now())) {
          auctionService.decideWinner(auction.getId());
          log.info(auction + " updated and closed!");
        }
      });
    } catch (Exception e) {
      log.error("DetermineAuctionWinnerJob exception: " + e.getMessage(), e);
    }
  }
}

Esas işi yapacak DetermineAuctionWinnerJob çalıştığında ihaleler tablosundaki tüm açık ihaleleri tarayarak, kazanan teklifi setliyor ve ihaleyi kapalı konumuna getiriyor.

Burada Job sınıfını yazarken, @Component annotasyonunu kullandık, çünkü Spring dependency injection ile AuctionService bağımlılığını sınıfa eklemek istedik.

Henüz job ile ilgili JobDetail ve Trigger sınıflarını yazmadık.Bu aşamada uygulamayı çalıştırdığımız zaman muhtemelen aşağıdaki Quartz log’larını göreceğiz.

2023-02-27T20:29:14.329+03:00  INFO 2612 --- [           main] org.quartz.core.SchedulerSignalerImpl    : Initialized Scheduler Signaller of type: class org.quartz.core.SchedulerSignalerImpl
2023-02-27T20:29:14.329+03:00  INFO 2612 --- [           main] org.quartz.core.QuartzScheduler          : Quartz Scheduler v.2.3.2 created.
2023-02-27T20:29:14.329+03:00  INFO 2612 --- [           main] o.s.s.quartz.LocalDataSourceJobStore     : Using db table-based data access locking (synchronization).
2023-02-27T20:29:14.330+03:00  INFO 2612 --- [           main] o.s.s.quartz.LocalDataSourceJobStore     : JobStoreCMT initialized.
2023-02-27T20:29:14.330+03:00  INFO 2612 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler meta-data: Quartz Scheduler (v2.3.2) 'demoapp' with instanceId 'Esats-MacBook-Air.local1677518954325'
  Scheduler class: 'org.quartz.core.QuartzScheduler' - running locally.
  NOT STARTED.
  Currently in standby mode.
  Number of jobs executed: 0
  Using thread pool 'org.quartz.simpl.SimpleThreadPool' - with 5 threads.
  Using job-store 'org.springframework.scheduling.quartz.LocalDataSourceJobStore' - which supports persistence. and is clustered.

2023-02-27T20:29:14.330+03:00  INFO 2612 --- [           main] org.quartz.impl.StdSchedulerFactory      : Quartz scheduler 'demoapp' initialized from an externally provided properties instance.
2023-02-27T20:29:14.330+03:00  INFO 2612 --- [           main] org.quartz.impl.StdSchedulerFactory      : Quartz scheduler version: 2.3.2
2023-02-27T20:29:14.330+03:00  INFO 2612 --- [           main] org.quartz.core.QuartzScheduler          : JobFactory set to: org.springframework.scheduling.quartz.SpringBeanJobFactory@28bc9749
2023-02-27T20:29:14.457+03:00  WARN 2612 --- [           main] JpaBaseConfiguration$JpaWebConfiguration : spring.jpa.open-in-view is enabled by default. Therefore, database queries may be performed during view rendering. Explicitly configure spring.jpa.open-in-view to disable this warning
2023-02-27T20:29:14.579+03:00  INFO 2612 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
2023-02-27T20:29:14.579+03:00  INFO 2612 --- [           main] o.s.s.quartz.SchedulerFactoryBean        : Starting Quartz Scheduler now
2023-02-27T20:29:14.600+03:00  INFO 2612 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler demoapp_$_Esats-MacBook-Air.local1677518954325 started.
2023-02-27T20:29:14.604+03:00  INFO 2612 --- [           main] c.k.quartzdemo.QuartzDemoApplication     : Started QuartzDemoApplication in 1.602 seconds (process running for 1.843)
2023-02-27T20:29:14.604+03:00  INFO 2612 --- [           main] c.k.quartzdemo.init.JobScheduleInit      : Application started

Şimdi de uygulama ayağa kalkarken Job’ı konfigüre edecek kodları yazıyoruz. Bunun için Spring tarafından sunulan CommandLineRunner interface’ini kullanacağız.

//init/JobScheduleInit.java
package com.kocesat.quartzdemo.init;

import com.kocesat.quartzdemo.job.DetermineAuctionWinnerJob;
import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.TimeZone;

@Component
@Slf4j
public class JobScheduleInit implements CommandLineRunner {

  private final Scheduler scheduler;

  public JobScheduleInit(Scheduler scheduler) {
    this.scheduler = scheduler;
  }

  @Override
  public void run(String... args) throws Exception {
    log.info("Application started");
    final JobKey jobKey = JobKey.jobKey("determineAuctionWinner", "demoApp");
    if (scheduler.checkExists(jobKey)) {
      scheduler.deleteJob(jobKey);
    }

    final JobDetail jobDetail = JobBuilder.newJob(DetermineAuctionWinnerJob.class)
      .withIdentity(jobKey.getName(), jobKey.getGroup())
      .withDescription("Determines the auction winner")
      .storeDurably()
      .build();

    final TriggerKey triggerKey = TriggerKey.triggerKey("determineAuctionWinnerTrigger", "demoApp");
    final CronTrigger cronTrigger = TriggerBuilder.newTrigger()
      .forJob(jobDetail)
      .withIdentity(triggerKey)
      .withSchedule(
        CronScheduleBuilder
          .cronSchedule("0/20 * * * * ?")
          .withMisfireHandlingInstructionFireAndProceed()
          .inTimeZone(TimeZone.getTimeZone("Europe/Istanbul"))
      )
      .build();

    scheduler.scheduleJob(jobDetail, cronTrigger);
  }
}

Dakikanın 0, 20, 40. saniyelerinde (Yani her 20 saniyede bir) çalışacak şekilde cron ifadesi yazıyoruz.

Uygulamayı tekrar çalıştırdığımızda Job’ın her 20 saniyede bir fire ettiğini yazdığımız log’lardan görebiliriz.

Job istenilen zamanlarda ayağa kalkıp görevini yerine getiriyor.

Şimdi bir restful servis yazıp, Job’ı durdurup başlatmayı sağlayacağız.

Job Durdur/Başlat

Bunun için öncelikle JobService adında bir servis yazacağız.

package com.kocesat.quartzdemo.service;

import com.kocesat.quartzdemo.exception.AppRuntimeException;
import com.kocesat.quartzdemo.exception.JobNotFoundException;
import org.quartz.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import static org.quartz.CronScheduleBuilder.cronSchedule;

@Service
@Transactional
public class JobService {
  private final Scheduler scheduler;

  public JobService(Scheduler scheduler) {
    this.scheduler = scheduler;
  }

  public String startStop(String jobName, String jobGroup) {
    final JobKey jobKey = JobKey.jobKey(jobName, jobGroup);
    try {
      if (!scheduler.checkExists(jobKey))
        throw new JobNotFoundException();
      var triggers = scheduler.getTriggersOfJob(jobKey);
      if (triggers.isEmpty()) {
        throw new JobNotFoundException("Trigger not found");
      }
      var trigger = scheduler.getTriggersOfJob(jobKey).get(0);
      Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
      if (triggerState == Trigger.TriggerState.PAUSED) {
        scheduler.resumeTrigger(trigger.getKey());
        return "Job resumed";
      } else {
        scheduler.pauseTrigger(trigger.getKey());
        return "Job paused";
      }
    } catch (SchedulerException e) {
      throw new AppRuntimeException("Job start stop exception");
    }
  }
}

Bu servisi kontrol etmek için ilgili Job’ın key’i ile çağırıldığında, öncelikle quartz tablolarından bu key ile tanımlı bir Job ve Trigger’ları olup olmadığını kontrol ediyoruz. Ardından, Trigger’ın statüsüne göre eğer halihazırda durdurulmuş bir Trigger ise scheduler.resumeTrigger() metodu ile devam ettirilmesini sağlıyoruz. TriggerState.PAUSED değilse scheduler.pauseTrigger() metodu ile durduruyoruz.

Bu kodu dışardan tetikleyebilmek için rest api JobController katmanını yazıyoruz.

// controller/JobController.java
package com.kocesat.quartzdemo.controller;
import com.kocesat.quartzdemo.service.JobService;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("job")
@RequiredArgsConstructor
public class JobController {
  private final JobService jobService;

  @PutMapping("/start-stop")
  public ResponseEntity<String> startStop(
    @RequestParam("name") String jobName,
    @RequestParam("group") String jobGroup)
  {
    return ResponseEntity.ok(jobService.startStop(jobName, jobGroup));
  }
}

Job’ların unique olarak tutulduğu key (Name ve group) ile ilgili Job’ı Postman üzerinden durdurup çalıştırabiliyoruz.

Job yeniden zamanlama (reschedule)

Şimdi de Job için default olarak belirlediğimiz ve uygulama ayağa kalkarken set edilen cron ifadesini dinamik olarak değiştirecek metotları yazalım.

JobService sınıfına bu işi yapabilecek yeni metodu, reschedule() ekliyoruz. Bu metot çağırıldığında yine aynı şekilde, ilgili Job ve Trigger’ları veri tabanından çekerek, eski Trigger’i yeni Trigger ile scheduler.reschedule(oldTriggerKey, newTrigger) kullanarak değiştiriyoruz.

// Add this code to /service/JobService.java  
  public String reschedule(String jobName, String jobGroup, String cronExpression) {
    final JobKey jobKey = JobKey.jobKey(jobName, jobGroup);
    try {
      if (!scheduler.checkExists(jobKey))
        throw new JobNotFoundException();
      var triggers = scheduler.getTriggersOfJob(jobKey);
      if (triggers.isEmpty()) {
        throw new JobNotFoundException("Trigger not found");
      }
      Trigger oldTrigger = triggers.get(0);
      Trigger newTrigger = oldTrigger
        .getTriggerBuilder()
        .withSchedule((ScheduleBuilder)cronSchedule(cronExpression))
        .build();

      scheduler.rescheduleJob(oldTrigger.getKey(), newTrigger);
      return "Job rescheduled";

    } catch (SchedulerException e) {
      throw new AppRuntimeException("Job start stop exception");
    }
  }

Rest api katmanında metodu kullanıma açıyoruz.

// Add this code to /controller/JobController.java  
  @PutMapping("/reschedule")
  public ResponseEntity<String> reschedule(
    @RequestParam("name") String jobName,
    @RequestParam("group") String jobGroup,
    @RequestParam("cron") String cronExpression
    )
  {
    return ResponseEntity.ok(jobService.reschedule(jobName, jobGroup, cronExpression));
  }

Postman’dan istek atarak yeni metodumuzu test ediyoruz.

Özet

İki bölümden oluşan bu yazımızda, Quartz kütüphanesinin temel bileşenlerinin üzerinden geçerek nasıl kullanıldığı öğrendik. Öğrendiklerimizi uygulayabilmek için de Spring Boot uygulaması ile mini bir ihale sistemi yazdık.

Akbank Teknoloji Ekibi olarak yazılım ve yeni teknolojilerle ilgili tüm deneyimlerimizi Medium yazılarımızla paylaşmaya devam edeceğiz. Bizi takip etmeyi unutmayın!

Kullanılan kodların tamamına GitHub (https://github.com/kocesat/quartz-demo) sayfasından ulaşabilirsiniz.

Referanslar

• http://www.quartz-scheduler.org/
• https://docs.spring.io/spring-boot/docs/2.0.0.M3/reference/html/boot-features-quartz.html
• https://docs.oracle.com/cd/E12058_01/doc/doc.1014/e12030/cron_expressions.htm
• https://www.baeldung.com/spring-quartz-schedule
• https://stackabuse.com/guide-to-quartz-with-spring-boot-job-scheduling-and-automation/