MySQL 8のSKIP LOCKEDを使って負荷のかからないジョブキューを実装する

MySQL 8で導入されたSKIP LOCKED

MySQL 8でSELECT ... FOR UPDATEにSKIP LOCKEDというオプションが加わった。これはロックされている行を除いた行をロックして取得できる。

ジョブキューに使える

公式ドキュメントにも記載されているとおり、queue-like table（つまりジョブキュー）に利用できる。

Queries that skip locked rows return an inconsistent view of the data. SKIP LOCKED is therefore not suitable for general transactional work. However, it may be used to avoid lock contention when multiple sessions access the same queue-like table.

同時実行時のパフォーマンス改善に使える

他には、チケットの先着順予約やセール品の在庫の取り合いなど、ロックが必要な同時実行処理のパフォーマンス改善に使える。

普通に設計するとチケット枚数や在庫数を数値型として一レコードに持つようにするところを、1枚1レコードとして枚数分の行をinsertしておき、SKIP LOCKEDつきのSELECT ... FOR UPDATEを投げることで、一レコードに対して多数のロック待ちが発生することがなくなる。

参考: 第123回　ロッキングリードのNOWAITとSKIP LOCKEDオプションについて

ジョブキューもパフォーマンスを悪化させずにMySQLで使える

RDBを使ってジョブキューを実装するのは、ポーリングやロック待ちによるスケーラビリティ、パフォーマンスの問題が発生するため避けるべきとされることが多い。

ただRedisを使うよりMySQLを使う方がデータの永続性の観点で安心だし、RabbitMQのようなMessageQueueをジョブキューのためだけに持ち出すのもヘビーなため、MySQLを使ってパフォーマンスの問題が出ないのであればMySQLを使いたい。

ジョブキューをSpring Bootの@Scheduledで実装してみる

簡単なジョブキューをSpring Bootの@Scheduledで実装してみる。デプロイするサーバは10台あり、10個のタスクが同時に動く可能性があるとする。

Application起動部、application.properties

Applicationクラスには、@Scheduledを有効にするため@EnableSchedulingをつけているだけ。

package com.example.skipLock;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@EnableScheduling
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}

propertiesファイルにはMySQLの接続情報を記載する。

# MySQL
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost/test
spring.datasource.username=root
spring.datasource.password=パスワード
spring.jpa.database=MYSQL
spring.jpa.hibernate.ddl-auto=update

Repository, Entity

データアクセス層にはSpring Data JPAを利用する。

Entityにはステータスがあるだけ。

package com.example.skipLock.entity;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

@Entity(name = "tasks")
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Task {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String status;
}

Repositoryでは、nativeクエリを使用して、SKIP LOCKEDオプションを手書きする。

package com.example.skipLock.repository;

import com.example.skipLock.entity.Task;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.CrudRepository;

import java.util.Optional;

public interface TaskRepository extends CrudRepository<Task, Long> {

    @Query(value = "select * from tasks where status = 'YET' order by id limit 1 for update skip locked", nativeQuery = true)
    Optional<Task> findYet();

}

以下の二つにより、登録した順で他のものにロックされていないものを1件だけ取得できる。

order by id
limit 1

SKIP LOCKEDを利用する際のジョブキューのステータス管理

条件部分のstatusはYETかCOMPLETEのどちらかの文字列が入る。

SKIP LOCKEDを用いたジョブキューのステータス管理に最低限必要なのは「未着手」と「完了」の二つだけで、「処理中」は必ずしも必要ではない。

SKIP LOCKEDを用いない通常のSELECT ... FOR UPDATEであれば、ロック待ちをできるだけ起こさないように、「未着手」のデータをロック付きで取得したらすぐに「処理中」に更新してロックを解放しなければならなかった。

しかしSKIP LOCKEDではロック待ちについて気にする必要がないため、処理しているあいだもロックを持ちっぱなしで問題ない。（もちろん、処理がものすごく長い場合は、「処理中」のステータスに変更した方がいいだろうが）

スケジュール処理

@Scheduledを使う。

SELECT ... FOR UPDATEを使うということは、トランザクション制御が必要なので、@Transactionalも忘れないようにしなければならない。

package com.example.skipLock.component;

import com.example.skipLock.entity.Task;
import com.example.skipLock.repository.TaskRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Optional;

@RequiredArgsConstructor
@Component
public class Scheduler {

    private final TaskRepository taskRepository;

    @Transactional
    @Scheduled(fixedDelay = 1000)
    public void task() {
        Optional<Task> task = taskRepository.findYet();
        task.ifPresent(x -> {
            // do something
            // ...

            x.setStatus("COMPLETE");
            taskRepository.save(x);
        });
    }
}

デバッガーを利用して動作確認

MySQLにアクセスして、適当な件数データを登録しておく。

mysql> insert into tasks (status) values('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET'),('YET');
Query OK, 18 rows affected (0.04 sec)
Records: 18  Duplicates: 0  Warnings: 0

Javaに戻って、SchedulerクラスのtaskRepository.save(x);にブレークポイントを置いてApplicationを起動する。

変数xを見ると、x.idが1であることがわかる。

ここでMySQLにコンソールから同じSQLを投げてみると、idが2のデータが取得でき、SKIP LOCKEDがうまく効いていることがわかる。

mysql> select * from tasks where status = 'YET' order by id limit 1 for update skip locked;
+----+--------+
| id | status |
+----+--------+
|  2 | YET    |
+----+--------+
1 row in set (0.00 sec)

環境

MySQL 8
Java 11
Spring Boot 2.5.2