TypeScriptで複数の非同期処理を制御する方法

TypeScriptで複数のリクエストを送る方法を試行錯誤し、ようやく良さげなのを見つけたのでまとめておきます。

※2020/10/8 直列処理でreduceを使わない方法を追加しました。

Promiseの使い方

複数のリクエストについて考える前に、まず単一リクエストの待ち合わせにPromise を用います。

main.ts

// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          console.log(`${item} done`);          
          resolve();
      }, 500 * item);
    })
  }
}

main() {
  //　非同期関数の呼び出しと待ち合わせ
  asyncFunc(3)
    .then(() => {
      console.log('OK');
    }).catch(() => {
      console.log('error');
    })
  console.log('main executed');

}

非同期関数は Promiseオブジェクトを返却します。オブジェクトの返却にはresolve(処理の成功時)、またはreject(処理の失敗時)を用います。
この関数を呼び出した側ではこの返却を待ち、then()(resolve時)、またはcatch()(reject時)でそれぞれの処理を実行することができます。

上記main関数を実行した場合のコンソール出力は以下の通りです。

asyncFunc3 called
main executed
3 done
OK

複数処理の実装例

上記Promiseの機能を踏まえた上で、以下のケースでの実装を考えます。
1. 並列処理
2. 順列処理

1. 並列処理

複数の処理を並列に実行し、全てが完了した後にその終了結果を受け取るというケースです。
これはPromise.allを用いることで複数のPromiseオブジェクトの返却を待つことができます。

main.ts

requestArray = [1,2,3,4,5];


// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          console.log(`${item} done`);          
          resolve();
      }, 500 * item);
    })
  }
}

main() {
    // 1
    const promises = [];
    this.requestArray.forEach((request) => {
      promises.push(this.asyncFunc(request));
    });

    // 2
    Promise.all(promises)
    .then(result => {
      console.log('all done');

    })
    .catch(err => {
      console.log('error happens');
    })

  console.log('main executed');

}

非同期関数は先ほどと同じものですが、今回は1回の呼び出しではなく、 requestArrayに入った要素の数ぶん呼び出しをします。
以下の手順で呼び出しと待ち合わせを実現しています。
1.非同期処理のオブジェクト配列生成
2.生成したオブジェクト配列を引数にPromise.all()実行

この場合のコンソール出力は以下の通りです。

asyncFunc1 called
asyncFunc2 called
asyncFunc3 called
asyncFunc4 called
asyncFunc5 called
main executed
1 done
2 done
3 done
4 done
5 done
all done

全ての非同期関数がまず呼び出され、その後全てが完了しオブジェクトの返却があったのち、

all done

が出力されています。

ちなみに複数並列にした場合のエラーの取り扱いですが、返却したいずれかのオブジェクトがrejectの場合にその時点でcatch()が実行されます。また、全てのオブジェクトから返却があったのちにthen()の処理は実行されません。

動きを見るために、非同期関数でエラーを発生させるよう修正をします。

main.ts

// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          if(item % 3) { // 追加: 3で割り切れる場合にはエラー扱い
            console.log(`${item} done`);          
            resolve();
          } else {
            console.log(`${item} error!`);
            reject();
          }
      }, 500 * item);
    })
  }
}

この場合のコンソール出力は以下の通りです。

asyncFunc1 called
asyncFunc2 called
asyncFunc3 called
asyncFunc4 called
asyncFunc5 called
main executed
1 done
2 done
3 error!
error happens
4 done
5 done

エラーが起きた3の時点でcatch()の処理が実行され、5まで終わってもthen()の処理は実行されません。

2. 順列処理

順列処理では、5つの非同期処理を1から順に、前の処理の完了を待って実行します。
この際には、 reduceを用います。
※2020/10/8追記: 後述の、 3.for-ofによる順列処理の方が良いかもしれませんので、参考程度にしてください

main.ts

// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          console.log(`${item} done`);          
          resolve();
      }, 500 * item);
    })
  }
}

main() {
    this.requestArray
    .reduce((promise, item) => {
      return promise
      .then(result => {})
      .catch(err => {console.log('func error');})
      .finally(() => {return this.asyncFunc(item);})
    }, Promise.resolve())
    .then(() => {
        console.log('all done');
      })
    .catch(() => {console.log('error happens');})
    .finally(() => {console.log('finally come');})
  console.log('main executed');

}

単純な順列処理だとひたすらthen()でつないでネスト地獄になる(しかも実行する数が静的でないと書きようがない)ところを、reduceが配列の要素数分実行してくれるので、すっきり書くことができます。

then()でつなぐネスト地獄の例.ts

asyncFunc(1)
  .then(() => asyncFunc(2)
                .then(()=> asyncFunc(3)
...

この場合のコンソール出力は以下の通りです。

main executed
asyncFunc1 called
1 done
asyncFunc2 called
2 done
asyncFunc3 called
3 done
asyncFunc4 called
4 done
asyncFunc5 called
5 done
all done
finally come

1つdoneになった後に次の呼び出しが実行されているのがわかります。

順列処理 - エラーでも最後まで処理を実行する

上記のコードで、非同期処理のどこかでエラーが起きた場合であっても中断することなく、5つの非同期処理を呼び出します。

ただし注意点があり、順列処理の最終的な成否は最後の非同期処理の成否に一致するということです。
つまり、今回の実装例では、5番目の処理の成否のみに応じて
Promise.resolve()).then(() => {成功}).catch(() => {失敗})
のいずれかに分岐するということです。

３番目の処理でエラーだったとしても、
5番目の処理が成功していれば、逐次処理としての最終結果は成功
という扱いになってしまいます。

この場合にエラーを拾う方法としては、非同期処理の中でエラー時のパラメータを別の変数に保持するというような実装にする他ないかもしれません。

main.ts

errorFlag = false; // 追加: エラーチェック用の変数

// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          if(item % 3) { // 追加: 3で割り切れる場合にはエラー扱い
            console.log(`${item} done`);          
            resolve();
          } else {
            console.log(`${item} error!`);
            this.errorFlag = true; // 追加：エラーの時にはフラグ変更
            reject();
          }
      }, 500 * item);
    })
  }
}

main() {
    this.requestArray
    .reduce((promise, item) => {
      return promise
      .then(result => {})
      .catch(err => {console.log('func error');})
      .finally(() => {return this.asyncFunc(item);})
    }, Promise.resolve())
    .then(() => {
        if(this.errorFlag) {             //修正: then()の中でエラーチェック用の変数に基づいて処理分岐
          console.log('error happens'); 
        } else {
          console.log('all done');
        }
      })
    .finally(() => {console.log('finally come');})
  console.log('main executed');

}

この場合のコンソール出力は以下の通りです。

main executed
asyncFunc1 called
1 done
asyncFunc2 called
2 done
asyncFunc3 called
3 error!
func error
asyncFunc4 called
4 done
asyncFunc5 called
5 done
error happens
finally come

最後まで実行されつつ、エラー自体も検知することができます。

順列処理 - エラー時点で後続処理キャンセル

エラーが発生しても最後まで実行する先の例よりも、こちらの方が実際の挙動としては必要なものかもしれません。

個人的に違和感がありますが、期待通りのことができたのは以下のような実装です。

main.ts

errorFlag = false;
// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          if(item % 3) {
            console.log(`${item} done`);          
            resolve();
          } else {
            console.log(`${item} error!`);
            this.errorFlag = true;
            reject();
          }
      }, 500 * item);
    })
  }
}

main() {
    this.requestArray
    .reduce((promise, item) => {
      return promise
      .then(result => {return this.asyncFunc(item);})  //修正: 成功時にのみ this.asyncFunc(item)を呼び出す
      .catch(err => {console.log('func error'); throw new Error(); }) //修正: エラー時に Errorを投げる
      .finally(() => {})
    }, Promise.resolve())
    .then(() => {console.log('all done')})
    .catch(() => {console.log(`got error`)})
    .finally(() => {console.log('finally come')})
  console.log('main executed');

}

この場合のコンソール出力は以下の通りです。

main executed
asyncFunc1 called
1 done
asyncFunc2 called
2 done
asyncFunc3 called
3 error!
func error
func error
func error
got error
finally come

3番目の処理(エラー発生)の時点でthrow new Error();としたことで、その後asyncFunc(4), asyncFunc(5)の呼び出しもなくエラーを返すかたちになりました。Promise.resolve())でもcatch()のエラー処理の方に分岐しています。

3.for-ofによる順列処理 2020/10/8追記

順列処理で良いのなら、単純にfor分による繰り返しで実装可能です。

main.ts

errorFlag = false;
// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          if(item % 3) {
            console.log(`${item} done`);          
            resolve();
          } else {
            console.log(`${item} error!`);
            this.errorFlag = true;
            reject();
          }
      }, 500 * item);
    })
  }
}

async main() { // awaitで非同期処理を実施するため、asyncをつける
    for(const item of this.requestArray) {
      await this.asyncFunc(item).catch(() => {console.log('error')});
    }
    console.log('all done');
    if(errorFlag) {
      console.log('error happens');
    }
}

この場合のコンソール出力は以下の通りです。

asyncFunc1 called
main executed
1 done
asyncFunc2 called
2 done
asyncFunc3 called
3 error!
error
asyncFunc4 called
4 done
asyncFunc5 called
5 done
all done
error happens

非同期処理の完了をawaitで待って、ループで呼び出すだけという非常に単純な作りになりました。reduceを使う必要はなさそうですね

順列処理 - エラー時点で後続処理キャンセル

main.ts

errorFlag = false;
// 非同期関数
function asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
          if(item % 3) {
            console.log(`${item} done`);          
            resolve();
          } else {
            console.log(`${item} error!`);
            this.errorFlag = true;
            reject();
          }
      }, 500 * item);
    })
  }
}

async main() { // awaitで非同期処理を実施するため、asyncをつける
    for(const item of this.requestArray) {
      if(this.errorFlag) { // エラーフラグが立っていたらloopを抜ける
        break;
      }
      await this.asyncFunc(item).catch(() => {console.log('error')});
    }
    console.log('all done');
    if(errorFlag) {
      console.log('error happens');
    }

この場合のコンソール出力は以下の通りです。

asyncFunc1 called
main executed
1 done
asyncFunc2 called
2 done
asyncFunc3 called
3 error!
error
all done
error happens

単純にforの中でエラーフラグをチェックするだけという単純な追記でできました。ソースも見通しやすいですね。

まとめ

ここまで挙げてきた非同期処理の実装をまとめたものをAngularベースのコードで載せておきます。
stackblitzはこちら ※2020/10/8 forでの実装を追記

app.component.ts

import { Component } from "@angular/core";
import { HttpClient } from "@angular/common/http";
@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  constructor(private http: HttpClient) {}
  requestArray = [1,2,3,4,5];
  respErrorArray = [];

  // force Error
  forceError = true; // change it when you want to cause error

  ngOnInit() {

    // 1 time
    // this.asyncFunc(3).then(() => {
    //   console.log('OK');
    // }).catch(() => {
    //   console.log('error');
    // })

    // pararell
    // this.promiseAllFunc();

    // sequential
    // this.reduceFunc();
    // this.reduceFuncErrorStop();

    // sequential with for-of loop
    // this.forLoop();
    this.forLoopErrorStop();
    console.log('main executed');

  }


  promiseAllFunc() {
    const promises = [];
    this.requestArray.forEach((request) => {
      promises.push(this.asyncFunc(request));
    });

    Promise.all(promises)
    .then(result => {
      console.log('all done');

    })
    .catch(err => {
      console.log('error happens on');
      this.respErrorArray.forEach((element) => {
        console.log(element);
      })  
    })
  }

  reduceFunc() {
    this.requestArray
    .reduce((promise, item) => {
      return promise
      .then(result => {})
      .catch(err => {console.log('func error');})
      .finally(() => {return this.asyncFunc(item);})
    }, Promise.resolve())
    .then(() => {
        console.log('all done');
        if(this.respErrorArray.length > 0) {
          console.log('error happens on');
          this.respErrorArray.forEach((element) => {
            console.log(element);
          })
        }
      })
    .finally(() => {console.log('finally come')})
  }
  async forLoop() {
    for(const item of this.requestArray) {
      await this.asyncFunc(item).catch(() => {console.log('error')});
    }
    console.log('all done');
    if(this.respErrorArray.length > 0) {
      console.log('error happens on');
      this.respErrorArray.forEach((element) => {
        console.log(element);
      })
    }
  }  

  async forLoopErrorStop() {
    for(const item of this.requestArray) {
      if(this.respErrorArray.length > 0) {
        break;
      }
      await this.asyncFunc(item).catch(err => {console.log('func error');});
    }
    console.log('all done');
    if(this.respErrorArray.length > 0) {
      console.log('error happens on');
      this.respErrorArray.forEach((element) => {
        console.log(element);
      })
    }
  }  

  reduceFuncErrorStop() {
    this.requestArray
    .reduce((promise, item) => {
      return promise
      .then(result => {return this.asyncFunc(item);})
      .catch(err => {console.log('func error'); throw new Error(); })
      .finally(() => {})
    }, Promise.resolve())
    .then(() => {console.log('all done')})
    .catch(() => {console.log(`got error`)})
    .finally(() => {console.log('finally come')})
  }  

  asyncFunc(item) {
    console.log(`asyncFunc${item} called`);
    return new Promise((resolve, reject) => {
      setTimeout(() => {
        if(this.forceError){
          if(item % 3) {
            console.log(`${item} done`);          
            resolve();
          } else {
            console.log(`${item} error!`);
            this.respErrorArray.push(item);
            reject();
          }
        } else {
          console.log(`${item} done`);          
          resolve();
        }

      }, 500 * item);
    })
  }

}

参考

• Promiseの配列を順に処理 | ryotah’s blog
• reduce関数は結構有用っていうお話 | あと味
• Promiseを返す関数の直列実行におけるreduceの利用と注意点 @berlysia | Qiita
• 2018-11-15 非同期処理の直列化：今やArray.reduceを使わなくてもできますよね | 銀の弾丸