예제들

https://www.jobrunr.io/en/blog/2020-04-23-jobrunr-long-running-jobs/

https://techfox.tistory.com/37

https://jsonobject.tistory.com/608

https://helpex.vn/article/jobrunr-kubernetes-terraform-6096552b1368de94091b8e65

https://github.com/jobrunr/jobrunr/pull/198

https://medium.com/edreams-odigeo-tech/email-scheduling-application-with-jobrunr-d37fd18b4898

# 중요

무료 버전도 클러스터를 지원한다.

각 클러스터들 끼리는 타임존이 맞아야 한다.

아래처럼 설정했다가 console log가 KST로 나오는 것을 봤고, 대시보드에서 9시간 차이나면서 jobrunr.server.ServerZooKeeper가 노드들 시간이 안맞는다고 죽이는 걸 봐서 main함수로 올려주었다.

시간 동기화는 NTP(Network Time Protocol)를 사용해서 한다고 하더라.

@SpringBootApplication
@EnableEurekaClient
//@MapperScan("com.fm.webinar.repository")
public class WebinarApplication {

    public static void main(String[] args) {
        TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
        SpringApplication.run(WebinarApplication.class, args);
    }

//    @PostConstruct
//    public void init() {
//        // Setting Spring Boot SetTimeZone
//        TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
//    }

}

예제가 거의 없어서 못찾았다.

그래서 일단 내 방식대로 띄운 다음에 이곳을 업데이트 하려고 한다.

pom.xml

<dependency>
    <groupId>org.jobrunr</groupId>
    <artifactId>jobrunr-spring-boot-starter</artifactId>
    <version>5.1.6</version>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-configuration-processor</artifactId>
    <optional>true</optional>
</dependency>

application.properties

## jobrunr start
jobrunr.datasource.jdbc-url=jdbc:mysql://localhost:3306/jobrunr?characterEncoding=UTF-8&allowMultiQueries=true
jobrunr.datasource.username=jobrunr
jobrunr.datasource.password=jobrunr
jobrunr.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
org.jobrunr.background-job-server.enabled=true
org.jobrunr.background-job-server.poll-interval-in-seconds=5
org.jobrunr.dashboard.enabled=true
org.jobrunr.dashboard.port=8000
org.jobrunr.database.type=sql
# dataSource bean name
org.jobrunr.database.datasource=jobrunrDataSource
org.jobrunr.database.skip-create=false
org.jobrunr.database.table-prefix=webinar_
## jobrunr end

hibernate(같이 사용하게 되면?)

spring.datasource.jdbc-url=jdbc:mysql://localhost:3306/mydb?characterEncoding=UTF-8&allowMultiQueries=true
spring.datasource.username=mydb
spring.datasource.password=mydb
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

config

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import javax.sql.DataSource;

@Slf4j
@RequiredArgsConstructor
@Configuration
public class JobRunrConfig {

    @Primary
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource dataSource() {
        return DataSourceBuilder.create().build();
    }

    @Bean(name = "jobrunrDataSource")
    @ConfigurationProperties(prefix = "jobrunr.datasource")
    public DataSource jobrunrDataSource() {
        return DataSourceBuilder.create().build();
    }

}

JobRunr 서비스

~~서비스를 같이 물고 있다보니 순환참조 오류가 나길래 static 처리하였다.~~

순환참조는 IoC 컨테이너 기능으로 처리가 가능하다.

job update는 pro 버전만 되는 것 같다.

import com.fm.webinar.domain.jobrunr.EventUtilEndRequestDto;
import com.fm.webinar.domain.message.base.BaseMessage;
import com.fm.webinar.domain.message.base.EventUtilMessage;
import com.fm.webinar.service.QuizService;
import com.fm.webinar.service.SurveyService;
import com.fm.webinar.service.VoteService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jobrunr.jobs.JobId;
import org.jobrunr.scheduling.BackgroundJob;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Instant;
import java.util.Optional;
import java.util.UUID;

@Slf4j
@RequiredArgsConstructor
@Component
public class RunrService {

    private final RedisTemplate redisTemplateHash;

    @PostConstruct
    public void init() {
        // redis
        hashOpsJobTemplate = redisTemplateHash.opsForHash();
    }

    private static HashOperations<String, EventUtilEndRequestDto, String> hashOpsJobTemplate;

    private static final String WEBINAR_JOB_INDEX = "webinar_event_util_job";


    private static void saveJob(EventUtilEndRequestDto requestDto, JobId jobId) {
        hashOpsJobTemplate.put(WEBINAR_JOB_INDEX, requestDto, jobId.asUUID().toString());
    }

    private static UUID getJob(EventUtilEndRequestDto requestDto) {
        Optional<String> oJobId = Optional.ofNullable(hashOpsJobTemplate.get(WEBINAR_JOB_INDEX, requestDto));
        if (oJobId.isPresent()) {
            return UUID.fromString(oJobId.get());
        } else {
            return null;
        }
    }

    public static void deleteJob(EventUtilEndRequestDto requestDto) {
        UUID oldJobId = getJob(requestDto);
        if (oldJobId != null) {
            try {
                BackgroundJob.delete(oldJobId);
            } catch (Exception e) {
                log.error("deleteJob error", e);
            }
        }
        deleteJobWithRedis(requestDto);
    }

    public static void deleteJobWithRedis(EventUtilEndRequestDto requestDto) {
        hashOpsJobTemplate.delete(WEBINAR_JOB_INDEX, requestDto);
    }

    public static void deleteJob(BaseMessage message) {
        EventUtilEndRequestDto requestDto = EventUtilEndRequestDto.builder()
                .sessionId(message.getSessionId())
                .id(message.getId())
                .topicType(message.getTopicType())
                .build();
        deleteJob(requestDto);
    }


    public static void messageEndScheduler(EventUtilMessage message) {
        // 스케줄 등록
        EventUtilEndRequestDto requestDto = EventUtilEndRequestDto.builder()
                .sessionId(message.getSessionId())
                .id(message.getId())
                .topicType(message.getTopicType())
                .build();

        // 있을 수도 있는 기존 잡 삭제
        deleteJob(requestDto);

        JobId jobId;

        switch (message.getTopicType()) {
            case QUIZ:
                jobId = BackgroundJob.<QuizService>schedule(Instant.now().plusSeconds(message.getSeconds()),
                        (x) -> x.setQuizEndMessage(requestDto));
                // 방금 등록한 잡 저장
                saveJob(requestDto, jobId);
                break;
            case VOTE:
                jobId = BackgroundJob.<VoteService>schedule(Instant.now().plusSeconds(message.getSeconds()),
                        (x) -> x.setVoteEndMessage(requestDto));
                // 방금 등록한 잡 저장
                saveJob(requestDto, jobId);
                break;
            case SURVEY:
                jobId = BackgroundJob.<SurveyService>schedule(Instant.now().plusSeconds(message.getSeconds()),
                        (x) -> x.setSurveyEndMessage(requestDto));
                // 방금 등록한 잡 저장
                saveJob(requestDto, jobId);
                break;
        }
    }

}

QuizService

vote와 survey도 quiz와 비슷하다.

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fm.webinar.domain.jobrunr.EventUtilEndRequestDto;
import com.fm.webinar.domain.message.base.TemplateStatusType;
import com.fm.webinar.domain.message.quiz.rate.QuizRateMessage;
import com.fm.webinar.domain.message.quiz.template.QuizTemplateMessage;
import com.fm.webinar.jobrunr.RunrService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.sql.Timestamp;
import java.text.MessageFormat;


@Slf4j
@RequiredArgsConstructor
@Service
public class QuizService {

    @Value("${spring.redis.topic.quiz.template}")
    private String quizTemplateTopic;
    @Value("${spring.redis.topic.quiz.answer}")
    private String quizAnswerTopic;
    private final RedisTemplate redisTemplate;

    private final RedisTemplate redisTemplateHash;
    private final ElasticsearchOperations elasticsearchOperations;

    private final ObjectMapper objectMapper;

    private final WhoamiService whoamiService;

    private final String INDEX_TEMPLATE_PREFIX = "quiz_";
    private final String INDEX_ANSWER_PREFIX = "quiz_answer_";

    @PostConstruct
    public void init() {
        hashOpsQuizRate = redisTemplate.opsForHash();
        hashOpsQuizRateCount = redisTemplate.opsForHash();
        hashOpsQuizTemplate = redisTemplateHash.opsForHash();
    }

    private HashOperations<String, String, QuizTemplateMessage> hashOpsQuizTemplate;
    private HashOperations<String, String, QuizRateMessage> hashOpsQuizRate;
    private HashOperations<String, String, Integer> hashOpsQuizRateCount;

    private final String QUIZ_RATE = "QUIZ_RATE";

    private final String QUIZ_RATE_COUNT = "QUIZ_RATE_COUNT";

    public void setQuizEndMessage(EventUtilEndRequestDto req) {
        QuizTemplateMessage message = getTemplate(req.getSessionId(), req.getId());
        QuizTemplateMessage endMessage = message.toBuilder().templateStatus(TemplateStatusType.END).build();
        saveTemplate(endMessage);
        redisTemplate.convertAndSend(quizTemplateTopic, endMessage);
        // 시간 내에 정상적으로 동작시 redis에 등록한 job은 삭제한다.
        RunrService.deleteJobWithRedis(req);
    }

    public void sendTemplate(QuizTemplateMessage message) {
        if (message.getTemplateStatus() == TemplateStatusType.START) {
            if (message.getSeconds() > 0) {
                message = message.toBuilder()
                        .endTime(new Timestamp(System.currentTimeMillis() + (message.getSeconds() * 1000L)))
                        .build();
            } else {
                // 지정된 시간이 없으면 1day로 설정한다.
                message = message.toBuilder()
                        .endTime(new Timestamp(System.currentTimeMillis() + (86400 * 1000L)))
                        .build();
            }

            RunrService.messageEndScheduler(message);
        } else if (message.getTemplateStatus() == TemplateStatusType.DELETE) {
            // 삭제 시 스케줄러 취소
            RunrService.deleteJob(message);
        }

        saveTemplate(message);
        redisTemplate.convertAndSend(quizTemplateTopic, message);
    }

    private String saveTemplate(QuizTemplateMessage message) {
        // redis
        hashOpsQuizTemplate.put(MessageFormat.format("{0}{1}", INDEX_TEMPLATE_PREFIX, message.getSessionId())
                , message.getId(), message);

        // es
        IndexQuery indexQuery = new IndexQueryBuilder()
                .withId(message.getId())
                .withObject(message)
                .build();
        return elasticsearchOperations
                .index(indexQuery, IndexCoordinates.of(INDEX_TEMPLATE_PREFIX + message.getSessionId().toLowerCase()));
    }

    public QuizTemplateMessage getTemplate(String sessionId, String id) {
        return hashOpsQuizTemplate.get(MessageFormat.format("{0}{1}", INDEX_TEMPLATE_PREFIX, sessionId), id);
    }


}

사용법

QuizService 중간에 보면 다음과 같이 잡을 등록하는 부분이 있다.

RunrService.messageEndScheduler(message);

위처럼 사용하면 잡이 등록되며 정상적으로 실행이 된다.

JobRunr 동적 스케줄러 dynamic scheduler job trigger