Springのバッチ処理をMySQLで排他制御する

複数サーバで稼働するSpringBootアプリケーションのバッチ処理をMySQLで排他制御したので備忘録として残します。

目的・前提

• 複数のサーバ上で同じSpringBootアプリケーションを稼働する。
• バッチ処理でDBのデータを操作するため排他制御する必要がある。
• バッチ処理は @crontab で起動する。

ライブラリ等

• spring-boot-starter: 2.6.4
• mybatis-spring-boot-starter: 2.2.2
• lombok: 1.18.22

結論

• MySQLのロック関数で排他制御が可能。
• ただしバッチ中にMySQLのセッションが変わることがあるので、 @Transactional で同一セッションであることを明示する。

実装

ロック関数を実行するRepository, Serviceを作成

• get_lock, release_lock 等をするServiceを作成する。

/src/main/java/{name-space}/repository/LockRepository.java

@Repository
public interface LockRepository {

    Integer isFreeLock(@Param("str") String str);

    Integer getLock(@Param("str") String str, @Param("timeout") int timeout);

    Integer releaseLock(@Param("str") String str);
}

/src/main/resource/mapper/LockRepository.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="{package-name}.repository.LockRepository">

  <select id="isFreeLock" resultType="Integer">
    select is_free_lock(#{str})
  </select>

  <select id="getLock" resultType="Integer">
    select get_lock(#{str}, #{timeout})
  </select>

  <select id="releaseLock" resultType="Integer">
    select release_lock(#{str})
  </select>
</mapper>

/src/main/java/{package-name}/service/LockService.java

@Service
@RequiredArgsConstructor
public class LockService {

    public enum LockResult {
        SUCCESS,
        FAILURE
    }

    private final LockRepository lockRepository;

    public boolean isFreeLock(String str) {
        Integer result = lockRepository.isFreeLock(str);
        if (result == null) {
            throw new ApplicationException("Database error");
        }
        return result == 1;
    }

    public LockResult getLock(String str, int timeout) {
        Integer result = lockRepository.getLock(str, timeout);
        if (result == null) {
            throw new ApplicationException("Database error");
        }
        return result == 1 ? LockResult.SUCCESS : LockResult.FAILURE;
    }

    public void releaseLock(String str) {
        Integer result = lockRepository.releaseLock(str);
        if (result == null) {
            throw new ApplicationException("Unknown lock");
        }
        if (result == 0) {
            throw new ApplicationException("Can not release lock");
        }
    }
}

バッチを作成

/src/main/java/{package-name}/batch/HourlyBatch.java

@Component
@RequiredArgsConstructor
public class HourlyBatch {

    private static final String LOCK_STR = "hourly_batch";
    private static final int LOCK_TIMEOUT = 0;

    private final LockService lockService;

    @Scheduled(cron = "0 0 * * * *")
    @Transactional
    public void execute() {
        // ロック確認
        try {
            LockResult lockResult = lockService.getLock(LOCK_STR, LOCK_TIMEOUT);
            if (lockResult == LockResult.FAILURE) {
                // 別プロセスがロック済み
                return;
            }
        } catch (Exception e) {
            // エラー処理
            return;
        }

        try {
            // データベース操作
        } finally {
            try {
                lockService.releaseLock(LOCK_STR);
            } catch (Exception e) {
                // エラー処理
            }
        }
    }
}