信じられない！何十億ものデータを ES に同期するのはとても簡単です。

1 これは背景です

最近、複数の条件を任意に組み合わせて注文データを照会する機能を提供してほしいという要望を受けました。データベース内の 1 億件を超える注文を見ると、髪の毛が 2 本抜け落ちたことがわかります。これは、この要件が単純ではないことを示しています。

写真

2 本の抜け毛の問題は、技術的に実装するのが難しくありません。実際、技術的な実装は明確かつ簡単です。データの異質性を介してデータを ES に同期し、ES の転置インデックス、キャッシュなどの機能を使用して、複数の条件で複雑なクエリを実行する機能を提供します。 ES クラスターはすでに存在します。

しかし、現在の ES インデックスには一部のデータが存在しません。つまり、1億件を超える注文データを注文データベースからESに再度更新する必要があり、この操作には1週間かかります。

何？信じられないなら、話し合いましょう。

2. 注文データをESに同期させる複雑さを見てみましょう

2.1 データ同期ESインデックスプロセス

写真

上図に示すように、これは ES インデックスにデータを同期するプロセスです。

まず、注文データベースからすべての注文データを照会し、次に注文データに保存されているユーザー ID、製品 ID、その他の情報に基づいて、ユーザーサービスと製品サービスから関連情報を照会する必要があります。処理および組み立て後、情報は ES クラスターに分類されます。

ユーザー情報と製品情報を照会する理由は、ES インデックス内の異種注文データが MySQL のデータと 1 対 1 で対応していないためです。商品カテゴリやユーザー情報などをもとに注文情報を照会したいという要望が多くあります。そのため、ここでは多くの上流サービスに照会して情報を収集する必要があります。

2.2 何か困難があるかどうかを整理してみましょう。

データベースから数億件の注文データを読み取ります。この操作はオンラインビジネスに影響を与えないため、照会される注文データベースは通常、ライブラリから取得されます。複数のデータソースを構成してデータを読み取ると、通常、数億件の注文が別々のライブラリとテーブルに保存されます。 16 個のライブラリに分割し、各ライブラリに 16 個のテーブル、合計 256 個のテーブルを用意しました。
何億もの注文データを一度にメモリに読み込むことはできません。そうしないと、メモリがいっぱいになり、煙が出てしまいます。したがって、ページングを考慮する必要があります。ページングに limit を直接使用するのは良くありません。データ量が増えると速度は遅くなります。したがって、カーソルを考慮する必要があります。さて、カーソルとしてフィールドを選択します。カーソルが一意で増加するのが最適です。
複数のサービスからデータを取得します。これらのデータが保存されているサービスは、通常、会社の他の部門に属しています。データを読み取るときは、他の人のサービスに影響を与えないようにしてください。あなたはここで「Ga Ga Meng」を照会しており、他の人のサービスが崩壊したのを見ると、非難されています。したがって、ここではフロー制御と分離を考慮する必要がありますね?完全なリンク分離は言うまでもなくコストが高すぎるため、少なくとも主要なサービスは分離する必要があります。
しばらくデータが同期された後、プロダクトマネージャーは、同期がどのくらいの期間行われているか、完了するまでにどのくらいの時間がかかるか、データの量はどれくらいあるかを尋ねました。私は混乱し、何も分かりませんでした。
途中で同期に失敗した場合はどうすればいいですか?再試行する必要がありますか?再試行するにはどうすればいいですか?再試行戦略とは何ですか?障害発生時にアラームは鳴りますか?時間内に検出して対処できますか?同期が一定期間中断された場合はどうなりますか?どこで中断されたかの記録はありますか?中断したところから同期を続行できますか?そうしないと、最初からやり直すのに N 日かかります。泣いてます。
一部を同期した後、問題が見つかったため、しばらく一時停止する必要があります。どうすればいいですか？
矛盾している注文データ、たとえば 2 つまたは 3 つの注文だけを同期したい場合は、どうすればよいでしょうか。注文 ID を手動で入力して ES データを同期する機能をまだ提供する必要がありますか?
同期プロセスはどのように機能しますか?開始時間は？終了時間は？合計でどれくらい時間がかかりますか？オペレーターは誰ですか？これらの統計はどこから来たのでしょうか?
深夜にデータを同期したいのですが、ビジネスにほとんど影響がない場合があります。夜に起きるためにアラームを設定できますか?
ここで、注文データだけでなく、製品 ES クラスターデータも同期する必要があります。このロジックをすべて書き直す必要がありますか?

ああ、考えただけで頭が痛くなる！

つまり、物事は単純に見えますが、実際にはそれほど単純ではないのです。

3 素晴らしいサービス

髪をもっと心地よくするために、上記の困難に対処する魔法のサービスが開発されました。それが ECP です。プロセス全体を自動化および視覚化できるため、ES への異種データのコストが削減されます。タスクインターフェースは次のとおりです。

写真

3.1 ECPの簡単な操作プロセス

簡単に言えば、ECP の役割は、データソースからデータを読み取り、それを ES 書き込みサービスにプッシュすることです。データ処理のロジックは業務ごとに異なるため、ES 書き込みサービスはさまざまなドッキングパーティによって実装され、単純なプロセスは次のようになります。

写真

これには、複数のデータソースからデータを読み取る方法、データソースの構成、SQL 検証、動的電流制限、SPI メカニズム、再試行戦略と障害認識、活性検出と障害回復、環境分離などの技術的な詳細が含まれます。

以下、順に紹介します。

3.2 複数のデータソースからのデータの読み取り

ECP は現在、ID ソース、テキストソース、スクリプトソースの 3 つのデータソースからのデータの読み取りをサポートしています。

3.2.1 IDソース

IDを入力するためのテキストボックスがあります。このシナリオは、小さなデータの同期に適しています。たとえば、一部のデータベースと ES データに不整合があることがわかった場合は、データを更新するだけで済みます。

写真

3.2.2 ファイルソース

ファイルソースとは、テキストファイルから取得されるデータソースを指し、中規模のデータの同期に適しています。 ECP はオブジェクトストレージに接続されており、ユーザーはオブジェクトストレージにファイルをアップロードできます。タスクが実行されると、ECP はオブジェクトストレージ内のテキストデータを読み取ります。

この場合、ユーザーがアップロードしたファイルは比較的大きい可能性があり、処理のためにメモリに直接読み込むのは現実的ではないことに注意してください。そのため、ここではストリーム方式を採用して読み取り、バッチを読み取り、バッチを処理し、バッチを解放することで、OOM が発生しないようにしています。

写真

簡略化された処理方法は次のとおりです。

 try (Response response = OK_HTTP_CLIENT.newCall(request).execute()) { if (!response.isSuccessful()) { throw new IOException("Unexpected code " + response); } // 以流的方式读取文件数据InputStream inputStream = response.body().byteStream(); BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)); }

3.2.3 スクリプトソース

スクリプトソースは、大量のデータを同期するのに適しています。

スクリプトは基本的に SQL とデータソースの組み合わせです。

ユーザーは ECP でデータベース接続情報を構成し、次に SQL を構成します。 ECP は SQL を実行し、構成されたデータベースからデータを読み取り、それを ES 書き込みサービスにプッシュします。

スクリプトソースは、数億のデータの読み取りとプッシュをサポートできます。次の図は、注文ライブラリ (サブライブラリとサブテーブル) に対して構成されたスクリプト情報を示しています。

写真

3.2.4 スクリプトソースビッグデータ読み取りの実装

数億のデータをメモリに読み込んで処理するのは明らかに不可能なので、ローカルデータを読み込んで処理するのが正しい方法です。

ビジネスでは、ページングがよく使用されます。ただし、ページングで制限オフセットとサイズのみを使用する場合、オフセット値が大きいとパフォーマンスが急激に低下し、SQL が遅くなり、データベース全体のパフォーマンスが低下することもあります。

したがって、ページ数が多い場合は、インデックス付きフィールドをカーソルとして指定する必要があります。このカーソルはページングのパフォーマンスを向上させることができます。たとえば、注文テーブルで注文 ID が増分され、インデックスが設定されている場合、SQL は次のように記述できます。select * from t_order where order_id > xxx order by order_id desc limit 10; order_id 値を変更するとページング効果が得られます。

この方法は良いのですが、ユーザーにカーソルインデックスの選択を要求すると、使用のハードルが間違いなく高くなります。したがって、ECP はビッグデータを読み込むために上記のページングメソッドを使用せず、次に示すように JDBC カーソルクエリメソッドを使用します。

 // 建立连接conn = DriverManager.getConnection(url, param.getDsUsername(), param.getDsPassword()); // 创建查询stmt = conn.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY, java.sql.ResultSet.CONCUR_READ_ONLY); stmt.setFetchSize(param.getFetchSize());

カーソルクエリは毎回 fetchSize 量のデータを読み取るため、大量のデータを読み取ることで発生する OOM の問題を効果的に回避できます。

3.3 SQLの解析と検証

ユーザーは SQL スクリプトを構成し、ECP は SQL スクリプトを検証して変更する必要があります。従来の文字列処理 (正規表現など) は、特定の状況では要件を満たすことができますが、エラーが発生しやすくなります。そのため、ECP は、SQL を AST 構文ツリーに解析して SQL に対してさまざまな処理を実行できる Druid の SQL 解析ツールキットを使用します。次の図に示すように:

写真

ECP によって提供されるデータサンプルクエリは、SQL に制限 1 を自動的に追加します。

写真

3.4 動的電流制限の実装

電流制限は、クラスター電流制限と単一マシン電流制限に分けられます。評価の結果、シンプルさを保つという原則に基づいて、単一マシンの電流制限を採用し、電流制限コンポーネントとして guava の RateLimiter を使用しました。

写真

ページで QPS 値が変更されると、その値はデータベースに同期されます。スケジュールタスクは値の変更を継続的にスキャンし、変更された値を RateLimiter コンポーネントに同期します。

もちろん、データ監視戦略（MQ のブロードキャストなど）を採用して、変更された値をよりタイムリーに RateLimiter に同期させることもできますが、この方法では他のコンポーネントの導入も必要となり、複雑さが急速に増大し、当社のシンプルな実装戦略に適合しません。

動的電流制限の実装プロセスは次のとおりです。

写真

次の図は、さまざまな時点で現在の制限値が変更された後の QPS の変化を示しています。

写真

3.5 再試行戦略と障害認識

ES と DB のデータは可能な限りリアルタイムの一貫性を確保する必要がありますが、最終的な一貫性も保証する必要があります。したがって、データのプッシュと処理が失敗した場合は、再試行を実行する必要があります。再試行するには？

まず、障害の種類を理解し、適切な再試行戦略を策定する必要があります。自分自身と敵を知ることで、あらゆる戦いで勝利が保証されます!

1. ネットワークジッタによるインターフェイス呼び出しのタイムアウト。マイクロサービス RPC インターフェイスを呼び出す場合、ネットワークジッターなどの条件により、インターフェイス呼び出しがタイムアウトする可能性がありますが、すぐに回復します。通常、これはたまにしか発生せず、次の通話では正常になります。

2. データ処理ロジックが異常です。この場合、異常は自動的に回復することはできず、手動で介入することしかできません。

3. 上流サービスの異常。アップストリームサービスの負荷が大きすぎてインターフェイス呼び出しが失敗した場合は、速度を落として処理を続行する必要があります。呼び出しを続けてアップストリームサービスをクラッシュさせることはできません。

上記の失敗タイプの特性と組み合わせると、フィボナッチ数列の再試行戦略は、フィボナッチ数列の特性（1、1、2、3、5、8、13、21、34、55、89…）に非常に適しています。

最初の失敗が発生すると、1 秒の遅延後に再試行が実行されます。タイムアウトの原因がネットワークジッタである場合、再試行は成功し、データ処理速度には影響しません。失敗の数が増えると再試行間隔が長くなり、上記の 2 番目と 3 番目の失敗の種類も考慮されます。

再試行コンポーネントは Guava Retry を使用します。簡単な疑似コードは次のとおりです。

 // 重试组件配置private final Retryer<Boolean> RETRYER = RetryerBuilder.<Boolean>newBuilder() // 对中断类的异常不重试.retryIfException(input -> !isPauseException(input)) // 1,1,2,3,5,8,13,21,33... .withWaitStrategy(WaitStrategies.fibonacciWait(1000, 30, TimeUnit.SECONDS)) // 重试次数达到一定的次数后，不再重试.withStopStrategy(StopStrategies.stopAfterAttempt(MAX_RETRY_TIMES)) .withRetryListener(new RetryListener() { @Override public <V> void onRetry(Attempt<V> attempt) { if (attempt.hasException()) { log.error("act=【DataFlushRpcCallRetry】desc=【重试】重试次数=【{}】重试异常=【{}】", attempt.getAttemptNumber(), attempt.getExceptionCause()); // 重试超过阈值进行报警提醒alarmIfExceedThreshold(attempt); } } }) .build(); // 将执行逻辑抽象为Runnable，对外暴露该方法public void execute(Runnable runnable) { innerExecute(runnable,RETRYER); } private void innerExecute(Runnable runnable, Retryer<Boolean> retryer) { try { retryer.call(() -> { runnable.run(); return true; }); } catch (Exception e) { log.error("act=【DataFlushRpcCallRetry】desc=【重试异常】error=【{}】", e); throw new IllegalStateException(e); } }

一定回数の再試行後に失敗した場合は、エラーメッセージがアラームグループに送信されます。プッシュされた情報に基づいて、エラーの種類、再試行回数、タスクの作成者などの情報を明確に把握できます。ほとんどの問題はログを確認しなくても見つけることができます。以下のように表示されます。

写真

3.6 処理のためにデータをどのサービスにプッシュする必要がありますか? -SPIメカニズム

ECP はユニバーサルサービスであるため、共通機能をまとめて完成品にし、非共通機能を抽象化してさまざまなドッキングパーティに引き渡して実装する必要があります。

シンプルな実装の観点から、サービスが ECP に接続したい場合は、ECP 上で開発し、サービスインターフェイスを呼び出して、データをサービスにプッシュします。アイデアは明確ですが、接続とメンテナンスのコストが非常に高く、統一された仕様がないため、お勧めできません。プロセスは次のとおりです。

写真

Java にはこの問題を解決できる優れたアイデアがあります。それが SPI です。そのため、ECP はインターフェースを提供して仕様を策定し、ES インデックスデータの具体的なアセンブリロジックは各ドッキングパーティによって実装されます。

このように、新しいドッキングパーティが参加する場合は、インターフェイスを実装するだけでよく、ECP は変更を加える必要がありません。

写真

サービス検出に関しては、ECP が採用している構成方法は、次の図に示すように、新しいタスクを作成するときにデータプッシュのコンシューマーサービスを選択することです。

写真

実装方法としては、同社が独自に開発したRPCフレームワークにより、呼び出しサービスを動的に指定する方式を提供している。疑似コードは次のとおりです。

 Reference<IEsIndexFlushAPI> reference = new Reference<>(); // 设置调用的服务名reference.setServiceName(serviceName); // 设置接口名reference.setInterfaceClass(IEsIndexFlushAPI.class); // 设置上下文reference.setApplicationConfig(applicationConfig); // 获取接口实例IEsIndexFlushAPI iEsIndexFlushAPI = ES_INDEX_FLUSH_API_MAP.computeIfAbsent(serviceName, s -> reference.refer()); // 接口调用log.info("act=【EsIndexFlushApiInvoker】desc=【请求值】serviceName=【{}】dataListSize=【{}】indexNameList=【{}】tag=【{}】", serviceName,request.getDataList().size(),request.getIndexNameList(),request.getTag() ); EMApiResult<FlushResponse> result = iEsIndexFlushAPI.flush(request);

3.7 環境隔離

データの同期は比較的負荷の高い操作ですが、オンラインビジネスに影響を与えることはありません。したがって、データを同期するためのサービスはオンラインサービスから分離する必要があります。 ECP は、アーキテクチャグループが提供するラベルルーティング機能を統合しており、リクエストリンク全体で指定されたラベルを持つサービスを呼び出して環境の分離を実現できます。

ECP ラベルルーティング構成図:

写真

次の図に示すように、ECP でタスクのラベルルーティングが FLUSH として設定されている場合、同期タスクの実行中に、リンク内の FLUSH ラベルにバインドされたサービスグループが自動的に呼び出されます。

写真

一部のサービスが FLUSH ラベルを持つグループとして構成されていない場合は、サービスの通常のオンライン環境が自動的に要求されます。このようにして、ある程度の環境隔離を実現できます。

写真

3.8 活性検出とタスク障害回復メカニズム

データプッシュプロセス中に予期せぬ事態が発生し、タスクが中断された場合はどうすればよいですか?

要件の期限を迎えた時、某年某月某日に進捗が1%になった時点でタスクが止まっていることに気づき、泣きました。

さらに、労働時間は短く、仕事量は多いです。中断がないかどうかを常にタスクに監視することはできませんよね?これは不適切かつ失礼です。

もちろん、ECP には「セルフレスキューパッケージ」があるため、このような状況は ECP では発生しません。次に、ECP のタスク検出と割り込み回復メカニズムについて説明します。

次の図に示すように、ECP には、活性検出とタスク障害回復という 2 つの主要コンポーネントがあります。活性検出コンポーネントは、現在のタスクスレッドの実行ステータスを監視する役割を担います。タスクスレッドが実行中の場合、タスクの生存時間が更新されます。タスク障害回復コンポーネントは、現在未完了のタスクをスキャンする役割を担います。タスクの最後の生存時間が指定されたしきい値よりも大きい場合、タスクは実行を再開するためにプルされます。

写真

更新の疑似コードは次のとおりです。

 @Scheduled(fixedDelay = ScheduleTimeConstants.KEEP_ALIVE_MILLS) public void renewal(){ futureMap.forEach((taskId,future)->{ if (!future.isDone()){ log.info("act=【renewal】desc=【任务续期】taskId=【{}】续期时间=【{}】",taskId, DateUtils.dateToString(new Date(),DateUtils.PATTERN)); contextService.renewal(taskId); }else { log.info("act=【renewal】desc=【任务结束】taskId=【{}】",taskId); futureMap.remove(taskId); } }); }

タスク障害回復の疑似コードは次のとおりです。

 @Scheduled(fixedDelay = ScheduleTimeConstants.RESTART_TASK_MILLS) public void restartTask(){ // 1.查询当前未完成的任务List<TaskFlushExecuteContextPO> contextPOS = contextService.queryRunningTask(); for (TaskFlushExecuteContextPO contextPO : contextPOS) { // 2.计算上次存活到当前的时间Integer durationMin = calculateTimeSinceLastAlive(); // 3.若时间大于指定阈值则对任务重新拉起if (durationMin >= MAX_DURATION_MIN){ log.info("act=【restartTask】desc=【任务重新拉起】taskId=【{}】",contextPO.getTaskId()); // 4.更新alive_time进行锁定防止并发执行int i = contextExtMapper.casUpdateAliveTime(); if (i >0){ // 5.重新拉起任务restart0(contextPO, aliveTime); } } } }

3.9 スムーズな移行の実現

通常、データを ES に同期する方法は 2 つあります。

データを元のインデックスに直接同期します。
新しいインデックスを作成し、二重書き込みとエイリアス切り替えを使用して、スムーズなトラフィック移行を実現します。

新しいインデックスを作成するときは、多くの場合、インデックスマッピングを変更したり、元のインデックスに影響を与えずに操作をロールバックできるようにしたりする必要があります。

このシナリオでは、ECP は過去に ES インデックスを手動で更新する手順を分析し、プロセスを抽象化して、次の図に示すように次の手順をまとめました。

写真

ECP は、Apollo 構成センターを統合してプッシュ機能を実装するスムーズな移行コンポーネントを提供します。簡単な実装プロセスは次のとおりです。

写真

3.10 エレガントなログ

次の図はタスク操作のログを示しています。原則として、ログレコードは非コア業務であり、コア業務コードから分離する必要があります。したがって、注釈付きのフローレコードを使用するのが適切な選択です。

写真

しかし、注釈付きフローレコードには問題があります。多くのシナリオでは、フロー内の値を動的に取得する必要があります。これは注釈を使用して実現できますか?答えはイエスです。上図に示すように、タスク ID とデータソースはどちらも動的データです。これを実現するにはどうすればよいでしょうか?次のコードを見てください。

 @Flow(subjectIdEp = "#taskPO.id",subjectType = SubjectTypeEnum.TASK,operateFlowType = OperateFlowTypeEnum.CREATE_TASK,content = "'创建任务，任务ID：' + #taskPO.id ") public void saveTaskWithUser(TaskPO taskPO) { String name = LoginUserContext.get().getName(); taskPO.setCreator(name); taskPO.setModifier(name); taskMapper.insertSelective(taskPO); }

subjectIdEp はフローサブジェクト ID であり、#taskPo.id は、パラメーター taskPo の ID 値を動的に取得するために使用できる式です。ここではspringEl式の能力が活用されています。

content = "'タスクの作成、タスク ID:' + #taskPO.id " はフロー情報であり、springEL 式はリクエストパラメーター taskPo の ID 情報を動的に取得するためにも使用されます。

ただし、一部の情報を取得するには、オブジェクトから単に値を取得するのではなく、一連の計算が必要になります。次のように：

 @Flow(subjectIdEp = "#contextPO.taskId", subjectType = SubjectTypeEnum.TASK, operateFlowType = OperateFlowTypeEnum.DATA_FLUSH, content = "'【数据同步】异常中断任务恢复执行，中断时间：' + T(com.zhuanzhuan.esmanage.utils.DateUtils).dateToStringSimple(#aliveTime)") @Transactional(rollbackFor = Exception.class,isolation = Isolation.REPEATABLE_READ) public void restart0(TaskFlushExecuteContextPO contextPO, Date aliveTime) { log.info("act=【restartTask】desc=【任务重新拉起】taskId=【{}】原aliveTime=【{}】", contextPO.getTaskId(), aliveTime); dsProcessorExecutor.executeAndKeepAliveMonitor(contextPO.getTaskId()); }

T(com.zhuanzhuan.esmanage.utils.DateUtils).dateToStringSimple(#aliveTime) は、DateUtils.dateToStringSimple メソッドが実行されることを意味し、これは、Spring コンテナーからオブジェクトを取得したり、オブジェクトのメソッドを呼び出したりするなど、式がメソッドを呼び出すことができることを意味します。

このアノテーションベースのパイプラインの実装原則は、SPEL 式と Spring Aop の特性を使用して、カスタムフローアノテーションをインターセプトするセクションを記述することです。疑似コードは次のとおりです。

 // 定义切面，拦截FLOW注解@Around("@annotation(com.zhuanzhuan.esmanage.entity.annotation.Flow)") public Object around(ProceedingJoinPoint point) throws Throwable { // 调用目标方法Object result = null; try { result = point.proceed(); recordFlow(point,result); return result; } catch (Throwable e) { recordException(point,e); throw e; } } // 流水记录的实现private void recordFlow(ProceedingJoinPoint point, Object result) { // try catch 防止影响主逻辑//TODO 看是否需要写在一个事务中，主要评估流水的重要性try { MethodSignature signature = (MethodSignature) point.getSignature(); Flow flowAnnotation = getFlowAnnotation(signature); // 组装参数上下文EvaluationContext evaluationContext = buildContext(point, signature); evaluationContext.setVariable("result",result); // ID表达式String subjectIdEp = flowAnnotation.subjectIdEp(); // content表达式String content = getContent(flowAnnotation, evaluationContext); // SPEL解析表达式Expression expression = PARSER.parseExpression(subjectIdEp); Integer subjectId = (Integer)expression.getValue(evaluationContext); record(flowAnnotation, subjectId, content); } catch (Exception e) { log.error("记录操作流水失败", e); } }