マイクロサービスにおける非同期処理は何で実現させるか？　Callback地獄などを起こさないKotlin Coroutineのメリット

自己紹介とセッションの概要

岩谷明氏：みなさんこんにちは。LINEの岩谷と申します。LINE Developers Meetup for Kotlinにご参加いただき、誠にありがとうございます。本日最初のセッションですが、「LINEでKotlinを活用してサービスを作っていく話」と題して、発表します。よろしくお願いいたします。

簡単に自己紹介します。私は2016年にLINE株式会社に入社しました。今までは「LINE LIVE」や「LINEノベル」などのiOSやAndroidのエンジニアをしていましたが、2019年から、本日お話しするライブコマースサービスのサーバーサイドエンジニアをしています。

（スライドを指して）本日お話ししたい内容はご覧のようになります。まず、今回私たちのチームが新しく開発した、ライブコマースサービスの紹介をさせてください。

次に、その開発に当たってどのような技術を利用したのか説明します。そして、今回の開発で導入したKotlin Coroutineの概要や、実際に利用してみて得られた知見について説明したいと思います。よろしくお願いします。

新サービス「LIVEBUY」の概要

それでは、新サービスについて紹介させてください。LINEのライブコマースサービス「LIVEBUY」が、2021年11月24日にテストローンチしました。いくつかのライブ配信をすでに行なっていて、ユーザーのみなさまはLINE上でライブ配信を見ながら、その場で商品が購入でき、リアルタイムなインタラクションを楽しんでもらえるサービスとなっています。

そもそもライブコマースとは何かというところも補足させてください。ライブコマースサービスは、（配信者は）ライブ配信をしながら商品を紹介して、ユーザーは配信を見ながら購入できるサービスです。

私たちは“セラー”と呼んでいますが、商品を販売する人は、ライブ配信をしながら商品を宣伝したり、紹介したりします。ユーザーは視聴しながら、商品について気になるポイントを直接チャットで質問したりして、セラーはそれにリアルタイムで答えることができます。

これによって、リアルなお店のようなコミュニケーションが行えて、スタティックなECサイトにはない買い物体験ができる。こういったサービスになります。

このサービスをLINE上で提供するために、今回はLINE Front-end Framework、通称LIFFを活用して作成しました。LIFFは、WebアプリケーションをLINE上で動かせるフレームワークになっています。LIFF自体はどなたでも作成できるので、ぜひ試してみてほしいと思います。

「LIVEBUY」はLIFF上で動くWebアプリケーションとして実装されました。（スライドを指して）これらのスクリーンショットのように、ライブを見ながらチャットや商品の注文が行えます。

また、セラー向けのCMSも開発しています。（スライドを指して）ここは先ほどお伝えしたような商品を販売するセラー向けのCMSになっていて、配信の登録や、商品の管理、注文や発送の処理までを行えます。

さらに、セラー向けにアプリも提供する予定になっています。スマートフォンのカメラを利用して、ライブ配信できるアプリです。ライブ中にユーザーからのチャットをチェックしたり、CMSのように配信の管理ができたりするアプリです。クロスプラットフォームでアプリを開発できるFlutterを利用しており、iOS・Androidで動作するアプリになっています。

使用した技術と選定理由

さっそくLIVEBUYの開発に選定した技術とその理由を説明します。（スライドを指して）この順番にいっていきたいと思います。

今回は、マイクロサービスとKubernetesを採用しています。ライブコマースでは、ライブ動画の配信と、商品の管理および決済というロジックのだいぶ異なる要件があります。

そのため、マイクロサービスに分離しやすいこと、またWebサイトやCMSなど、ユーザーがアクセスするコンポーネントが多岐にわたること。また、モノリスのつらさなどの経験から、マイクロサービスの推奨より開発人数はだいぶ少ないですが、メリットがあると判断して採用しています。

そして、このマイクロサービスのオーケストレーションのために、Kubernetesを導入して、スケーラビリティを担保しています。社内にはマネージドなKubernetesクラスタを運用する専門のチームがいるため、アプリケーション開発者としては、Kubernetesクラスタを運用するつらさはあまりありません。

そして、今回のメインテーマとなるKotlinです。会社としてはもともとJavaをメインに利用しているため、その他のJVM（Java Virtual Machine）言語への興味・関心も非常に多い会社になっています。今回はKotlinの書きやすさと、Kotlin Coroutineによる非同期処理との親和性に注目して採用しています。

また、gRPCも利用しています。複数コンポーネントの通信が多いこと、また、複数の開発チームをまたいだスキーマの管理のためというのが採用のメインの理由となっています。チームでは、Server streaming RPC処理を利用できそうな要件もあったため、そこでも利用しています。

その他、リレーショナルデータベースとしてはMySQL、キーバリューストアとしてはRedis、イベントのストリーミングに関してはKafkaを利用しています。このあたりは社内でほぼ必ず利用するテックスタックになっていて、それぞれ専門に面倒を見てくれるチームが存在しています。

現在動作しているアーキテクチャ

これらのテックスタックを用いて、実際に構成した現在も動作しているアーキテクチャについて、説明したいと思います。（スライドを指して）上から見ていくと、まずユーザーが利用する各種のWebサービスやアプリケーションがあり、WebサービスはgRPC-Web、アプリはgRPCで私たちのサービスに通信してきます。

これらのリクエストはLoad Balancerを通り、Kubernetesのノードに振り分けられます。さらに、Kubernetes内ではenvoy、DNS roundrobinによって実際のポッドに振り分けられます。この際に、gRPC-WebとgRPCの変換なども行なっています。

また、社内のRest APIでJSONで通信してくるようなサービスからのリクエストも、JSON-gRPCトランスコーダを利用してgRPCに変換しています。そして、ユーザーからのアクセスを担うゲートウェイ相当のサービスにリクエストが届き、そこからさらに裏側のドメインを管理する各サービスを呼び出しています。

ここで、このマイクロサービスにおいてどうパフォーマンスを出すのかについて注目します。マイクロサービスではたくさんのRPC（Remote Procedure Call）を行い、通信が発生します。この通信では、自サービスのKubernetes pod間の通信もありますし、社内の別サービスの通信も多いです。

これらのRPCを同期的に実装してしまうと、通信にかかる間のThreadはブロックしてしまうため、とても無駄が多くなってしまいます。そのため、マイクロサービスではパフォーマンスを出すために、できるだけ非同期で通信を行いたいです。

「できるだけ」というのは、何かリクエストを受け取ってから、必要に応じてデータベースや別のサービスから情報を取得してレスポンスを返すまで、ということです。

今まで非同期処理といえば、Futureを利用したり、RxJavaやReactorなどのリアクティブプログラミング用のフレームワークを利用したと思いますが、callback地獄や、リアクティブプログラミングについてまわる難しさなどの欠点もありました。

Kotlin Coroutine

そこで、今回採用したKotlin Coroutineが登場します。Coroutineは中断可能な処理のことで、Threadに似ていますが、より軽量でコンテキストスイッチのコストもない概念になっています。また、特定のThreadには結びついてもいません。

ここで、簡単な例を見ていきたいと思います。（スライドを指して）まず上から見ていくと、launchというCoroutineビルダーが登場します。この中に書いた処理が、いわゆるCoroutineとなって独立して並行に動作します。

この中でdelayというSuspending functionを呼び出し、Coroutineは1秒サスペンドされて動作が止まります。しかし、動作が止まるといっても、Threadはブロックしないので、その他のCoroutineやコードは動作します。

そのため、下にある「Hello」が先に出力されます。1秒たった後に、Coroutineがサスペンドから復帰するので「World！」が出力される流れになります。

メイン関数の全体を囲っているrunBlockingは、通常の同期的なコードからCoroutineを起動する特別なfunctionになっていて、Coroutine内の処理が完了するまでは、起動されたThreadをブロックします。そのため、この後、プログラムは「Hello World！」と全体を出力するまで終了しないプログラムになります。

先ほど「delayはSuspending functionだ」と説明しましたが、補足をさせてください。Suspend functionというのは、Suspending修飾子をfunctionにつけた関数のことです。

Coroutineビルダーによって作られるCoroutineの中をCoroutine Scopeと言いますが、このCoroutine Scope内か別のSuspending functionからのみ、Suspending functionを呼び出せます。そのため、できる限りの処理を非同期にして実際にコーディングをすると、ほとんどSuspending functionを書くことになります。

もう1つ、Coroutineの重要な概念である、Structured Concurrencyについて説明します。本日は時間の関係で説明できませんが、エラー処理などにも関係する重要な概念になっています。

Coroutineは別のCoroutineからしか起動できない仕組みになっていて、Coroutineを起動した外側のCoroutineは、内側のCoroutineすべてが完了するまで完了しません。これが、Structured Concurrencyという概念になります。

勝手にずっと動いてるようなCoroutineは作れますが、面倒が見れないのと、リソースのリークにもつながるので、できるだけ作らないほうがよいとされています。

Suspending function内では、Coroutine Scopeという仕組みで親子構造を作れます。（スライドを指して）この例だと、doWorldというfunction内でlaunch functionにより並列に動作する2つのCoroutineが作成され、それぞれがCoroutine Scopeになります。この親のCoroutine Scopeでは最後に「Hello」を出力しており、launch内ではdelayを呼んでいるため、最初に出力されます。

しかし、Structured ConcurrencyによってdoWorldはこれでは終了せず、「World 1」「World 2」が出力されるまで終了されないことになります。最後に、doWorldが終了した後に呼び出し側の「Done」が出力され、全体が終わる流れになります。

以上がKotlin Coroutineの簡単な説明になります。これを実際どう利用していくか説明したいと思います。

Kotlin Coroutineの利用方法

Spring Framework5.2.0でCoroutineのサポートが入り、ControllerアノテーションをつけているControllerクラスでレスポンスを返すfunctionを、Suspending functionとして書けるようになったり、Coroutineの実行結果を表すDeferredというクラスがありますが、これをそのまま戻り値として利用できるようになったりしています。

Springでは非同期処理全般にReactorを利用しているので、いろいろなReactorをCoroutineに変換することで、さまざまな処理をCoroutineで書けるようになります。この変換は公式でできるようになっていて、（スライドを指して）ここに示すパッケージを利用することで可能になります。

今回gRPCには、gRPC-javaとgrpc-spring-boot-starterを利用しています。この例では、FooGrpcServicesというものが定義されていて、getBarというgRPCによって生成されたfunctionがあります。

このレスポンスを返すのを非同期にするため、Coroutine内で処理を行いたいので、Coroutineを起動できるように、別のcoroutineLancherというクラスを開発して利用しています。このクラスについて説明したいと思います。

このクラスでやっていることは、引数で受け取ったブロックをlaunch内で実行して、レスポンスを組み立てて、gRPCのresponceObserverを非同期に呼び出している処理です。さらにCoroutineExceptionHandlerというエラーハンドラをCoroutine Scopeに設定して、例外が発生した際はresponceObserverにエラーを返す処理になっています。

続いて、RedisをCoroutineから非同期で扱う方法を紹介します。spring-boot-starter-data-redis-reactiveを利用すると、非同期でRedisでのアクセスを行えます。（スライドを指して）ここにエクステンションとしてSuspending functionが追加されているので、最終的にKotlin Coroutineで書ける例になっています。

続いて、データベースへの非同期アクセスについてはR2DBC（Reactive Relational Database Connectivity）という仕様があり、Spring Data R2DBCを利用することで可能になります。近年、MySQL用のドライバが出てきたので、MySQLにもR2DBCで接続できるようになりました。

導入した感想・知見

導入してみた感想や知見を共有したいと思います。まずKotlin Coroutineですが、利点として、難しくなりがちな非同期処理を簡潔に書けて、コードの見通しが良くなることがメリットだなと思っています。とにかくSuspending functionで書けば、非同期なコードを書いている意識もあまりなく、書きやすく非同期処理を実行できます。

デメリットとしては、まだIDE（Integrated Development Environment）のサポートが追いついていないことや、Structured Concurrencyの概念が、ややとっつきにくいことが挙げられると思います。

gRPCを導入してみた感想としては、JSONの定義で消耗しなくてよくなったのがすごくいい利点だなと思いました。サーバー、アプリ、フロントなど、担当者が異なっていますが、齟齬が起きにくかったり、使い回しができたりしています。

マイクロサービスとKubernetesについてです。まずはマイクロサービスについてですが、個人的に感じてるのは、ドメインごとの境界のインターフェースがgRPCとして定められて明確になるので、結果としてコードがきれいにメンテしやすくなると感じています。

一方でマイクロサービスでは、通信先のサービスが死んでも影響がないように作る必要があったりと、エラーハンドリングがとても重要になってくるため、しっかり意識しないといけないと思います。

Kubernetesについては、やはり独自の概念がとっつきづらいので、学習曲線は高めだと思いますが、YAMLにすべての設定が書いてあるので、読み込めるようになると便利だと感じています。

また、R2DBCについては、今までデータベース、特にMySQLは非同期にしづらいと言われてきましたが、今回MySQLドライバが登場したことで導入でき、とてもいい時代になったなと感じています。一方で、Spring内ではReactorが利用されているので、そこである程度慣れないと、エラーが起きた時などに調査が難しいと思います。