redux-sagaで排他制御をするサンプル

const initialState = {
  // このデータを複数のsagaが取り合う
  items: ["a", "b", "c"],
}

export default function reducer(state = initialState, action) {
  switch (action.type) {
    // 指定されたデータをstoreから削除
    case "REMOVE": {
      const { item } = action
      return {
        items: state.items.filter(x => x !== item),
      }
    }
    default:
      return state
  }
}

import { fork, select, put, join } from "redux-saga/effects"

function* popItem() {
  // リストを取得
  const items = yield select(state => state.items)
  if (items.length <= 0) {
    throw new Error("insufficient items")
  }

  // リストから先頭のデータを選択
  const item = items[0]

  // 選択したデータをリストから削除
  yield put({ type: "REMOVE", item })

  // 呼び出し元に選択したデータを返す
  return item
}

export default function* mainSaga() {
  // 3個のsagaを起動
  const tasks = [yield fork(popItem), yield fork(popItem), yield fork(popItem)]

  // sagaの終了を待つ
  yield join(tasks)

  // 各sagaが取得したデータを表示
  console.info(
    "fetched items =",
    tasks.map(x => x.result())
  )
}

% yarn start
yarn run v1.21.1
$ babel-node src/main.js
store changed = { items: [ 'b', 'c' ] }
store changed = { items: [ 'b', 'c' ] }
store changed = { items: [ 'b', 'c' ] }
fetched items = [ 'a', 'a', 'a' ] # すべてのsagaが 'a' を取得してしまった!
Done in 0.55s.

import { call, fork, select, put, join } from "redux-saga/effects"
import { Semaphore } from "await-semaphore"

// セマフォ作成
// 引数が 1 ならば同時にロックを獲得できる saga は1つ
// 引数 1 を指定するなら, new Mutex() と等価
const sem = new Semaphore(1)

function* popItemWithLock() {
  // sem.acquire を呼び出して, ロックの獲得をする
  // もし, 他の saga がロックを獲得していれば, その saga が release() を呼び出すまで待つ
  // 他の saga がロックを獲得していなければ, すぐにロックを獲得できる
  //
  // call([sem, sem.acquire]) は sem.acquire() の呼び出しを意味する
  // https://redux-saga.js.org/docs/api/#callcontext-fn-args
  //
  // sem.acquire() は Promise を返すため, yield call でロックの獲得を待つ
  // https://www.npmjs.com/package/await-semaphore#semaphoreacquire-promise--void
  const release = yield call([sem, sem.acquire])
  try {
    // このブロックを実行できるsagaは高々1個
    const items = yield select(state => state.items)
    if (items.length <= 0) {
      throw new Error("insufficient items")
    }
    const item = items[0]
    yield put({ type: "REMOVE", item })
    return item
  } finally {
    // 忘れずにロックを解放する
    release()
  }
}

export default function* mainSagaWithLock() {
  const tasks = [
    yield fork(popItemWithLock),
    yield fork(popItemWithLock),
    yield fork(popItemWithLock),
  ]
  yield join(tasks)

  console.info(
    "fetched items =",
    tasks.map(x => x.result())
  )
}

% yarn start --enable-lock # オプションをつけると mainSagaWithLock が起動する
yarn run v1.21.1
$ babel-node src/main.js --enable-lock
store changed = { items: [ 'b', 'c' ] } # 'a' が取り出される
store changed = { items: [ 'c' ] }      # 'b' が取り出される
store changed = { items: [] }           # 'c' が取り出される
fetched items = [ 'a', 'b', 'c' ] # 相異なるデータを取得できた
Done in 0.53s.