출처글

querydsl paging reader thread-safe

multi thread를 사용하려면 thread safe가 되어 있는 Reader와 Writer를 사용해야하는데. 내가 사용한 querydslReader와 Writer는 safe가 맞다는 인증한 글을 참고하였다.

🔨batch의 multi thread 설정

batch에서의 multi thread 환경 설정은 excutor를 사용하면 된다.

    //로직 실행시 한번에 읽어올 데이터 (=chunk의 크기 == transaction 크기)
    private final int CHUNK_SIZE = 1;
    private final int POOL_SIZE = 2;   //multi thread
    private long totalBeforeTime = 0;
    
    @Bean
    public TaskExecutor orderDivisionExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(POOL_SIZE);    //Thread Pool 기본 사이즈, 최초 세팅 사이즈
        executor.setMaxPoolSize(POOL_SIZE);     //Thread Pool 최대 사이즈
        executor.setThreadNamePrefix("order-division-thread-");
        executor.setWaitForTasksToCompleteOnShutdown(Boolean.TRUE); //작업이 종료되면 Thread도 종료
        executor.initialize();
        return executor;
    }

job을 지정하는 곳에 다음과 같이 Bean으로 TaskExecutor를 만들고 POOL_SIZE는 상수로 원하는 크기만큼으로 설정해준다.

    @Bean
    @JobScope
    public Step orderProcessorStep() {
        return stepBuilderFactory.get("orderProcessorStep")
                .transactionManager(pt)
                .<Long, Long>chunk(CHUNK_SIZE)
                .reader(orderProcessorReader())
                .writer(orderProcessorWriter())
                .taskExecutor(orderDivisionExecutor())	//task 설정
                .throttleLimit(POOL_SIZE)   //생성된 Thread 중 사용할 Thread의 개수
                .build();
    }

그리고 job을 만들 때 writer 뒤에 해당 설정을 해준다.

그리고 중요한것은 mulit thread로 batch를 작동시키면 batch에 장점 중 하나인 실행시 실패했을 경우 실패 부분부터 재시작하는 것을 할수 없게된다.
그 이유는 비동기로 작업이 진행되다 보니 실패 시점을 알아도 이후의 작업들이 정상 처리되었을지 안되었을지 명확하게 알 수 없기 때문이다.
그래서 우리는 Reader에서 saveState를 false로 설정해주어야한다.
따로 설정을 안해도 되는지는 모르겠지만 해보자 한번

QuerydslPagingItemReader

QuerydslPagingItemReader의 메서드 중에 save option을 제거하고 생성하는 생성자가 없더라 그래서 상속 받고 있는 AbstractPagingItemReader 내부를 살펴봤다.

그런데 AbstractPagingItemReader 내부에도 따로 설정하는 옵션이 없더라 그래서 또 AbstractItemCountingItemStreamItemReader를 살펴보니 그제서야 saveState가 보였다.

기본 값이 true인걸로 보아 우리가 false로 바꿔줘야할거 같다.

	public void setSaveState(boolean saveState) {
		this.saveState = saveState;
	}

	/**
	 * The flag that determines whether to save internal state for restarts.
	 * @return true if the flag was set
	 */
	public boolean isSaveState() {
		return saveState;
	}

구현되어있는 메서드 중 setSaveState()를 사용하여 값을 false로 바꿔서 생성하는 생성자를 구현해보았다.

    //multi thread 사용시 save option 제거
    public QuerydslPagingItemReader(EntityManagerFactory entityManagerFactory,
                                    int pageSize,
                                    int state,
                                    Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        this();
        this.entityManagerFactory = entityManagerFactory;
        this.queryFunction = queryFunction;
        setPageSize(pageSize);
        setSaveState(state == 1 ? true : false);
    }

기존의 transacted 설정하는 생성자가 존재해서 boolean 값을 그대로 사용하진 못하고 그렇다고 매개변수의 순서를 바꾸자니 나중에 더 헷갈릴거 같고 그렇다고 또 enum으로 정의해놓자니 true false만 구분하면 되기 때문에 Integer의 0과 1로 구분하도록 했다.

    @Bean
    @StepScope
    public QuerydslPagingItemReader<? extends Long> orderProcessorReader() {
        return new QuerydslPagingItemReader<>(emf, CHUNK_SIZE, 0, queryFactory ->
                queryFactory
                        .select(order.idx)
                        .from(order)
                        .where(order.payWay.eq(8).and(order.ordStep.lt("200")))
        );
    }

그리고 생성할 때 매개변수로 0을 전달하면 false로 세팅된 Reader를 생성할 수 있게되었다.

QuerydslNoOffsetPagingItemReader

QuerydslNoOffsetPagingItemReader은 위 QuerydslPagingItemReader를 상속하여 생성하고 있기 때문에 위에서 만든 생성자로 만드는 생성자를 또 추가해주면 된다.

    //multi thread 환경일 경우 saveState(false)
    public QuerydslNoOffsetPagingItemReader(EntityManagerFactory entityManagerFactory,
                                            int pageSize,
                                            QuerydslNoOffsetOptions<T> options,
                                            int state,
                                            Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        super(entityManagerFactory, pageSize, state, queryFunction);
        setName(ClassUtils.getShortName(QuerydslNoOffsetPagingItemReader.class));
        this.options = options;
    }

다음과 같이 동일하게 작성해주고

    @Bean
    @StepScope
    public QuerydslNoOffsetPagingItemReader<OrderDivision> orderDivisionSimpleReader() {

        QuerydslNoOffsetNumberOptions<OrderDivision, Long> options = new QuerydslNoOffsetNumberOptions<>(orderDivision.idx, Expression.ASC);

        return new QuerydslNoOffsetPagingItemReader<>(emf, CHUNK_SIZE, options, 0, queryFactory ->
               ...
        );
    }

다음과 같이 작성했다.

Multi Thread 환경으로 batch를 돌리는것을 권장하는 것 같지는 않으나 회사 업무 차원에서 대용량이의 데이터를 정해진 시간 내에 최대한 빠르게 많은 작업을 해야할 일이 있어서 사용하게 되었다. 이동욱님도 해당 기능을 도입할 땐 최대한 고민해보라고 하시는걸 보면 정말 빠르게 처리해야할 경우에만 도입하는게 좋을거 같다!