スクリプト開発 / コネクターサブスクリプション

一部のコネクターはメッセージのサブスクライブをサポートしており、DataFlux Func はサブスクライブのための統一された方法を提供しています。

以前のバージョンでは「データソース」と呼ばれていましたが、現在は「コネクター」に変更されています

1. はじめに

DataFlux Func のスクリプト実行メカニズムにより、関数は起動後に最終的には必ず終了する必要があり、関数を無限に実行することは許可されていません。

そのため、サブスクライブのような長期間常駐する処理は、DataFlux Func 内でコンシューマーを直接作成して長時間実行する方法はサポートされていません。

代わりに、コネクターでサブスクライブするトピックを指定し、DataFlux Func メインプログラムがメッセージのサブスクライブを一元的に担当します。

DataFlux Func メインプログラムがメッセージを受信すると、指定されたメッセージ処理関数にメッセージを転送して処理を実行し、サブスクライブメッセージの処理を完了します。

2. サブスクライブをサポートするコネクター

最新版 DataFlux Func は、以下のコネクターのサブスクライブをサポートしています

3. 操作手順

Redis メッセージをサブスクライブして処理する場合の具体的な操作手順は以下のとおりです：

3.1 メッセージ処理関数の作成

メッセージ処理関数は固定の関数形式を持ちます。例：

Python
@DFF.API('Message Handler')
def message_handler(topic, message):
    print('topic', topic)     # トピック
    print('message', message) # 内容

スクリプトを完成したら、保存して公開します：

3.2 コネクターの作成と設定

基本設定の入力に加えて、サブスクライブするトピック（Topic）と対応する処理関数を追加で入力する必要があります。

処理関数は、前述の message_handler(...) 関数です：

3.3 メッセージのパブリッシュと処理の確認

パブリッシャーが以下のようなメッセージをパブリッシュすると、メッセージは DataFlux Func メインプログラムを経由して前述の message_handler(...) 関数に転送され、処理されます：

コネクター設定ページでは、対応するトピックの下に最新の消費情報が表示されます：

クリックすると、より詳細なタスク情報を確認できます：

これにより、Redis にパブリッシュされたメッセージが確かに message_handler(...) 関数によって受信され、処理されたことを確認できます。

4. メッセージのパブリッシュ

サブスクライブをサポートするコネクターは、一般的にメッセージのパブリッシュ（publish）もサポートしています。

メッセージのパブリッシュについては、スクリプト開発 / コネクターオブジェクト DFF.CONN 内の対応するコネクターオブジェクトの API を参照してください。通常は .publish(topic, message) の形式です。

5. メッセージ処理関数のタスク記録

最新版 DataFlux Func旧版 DataFlux Func

最新版の DataFlux Func では、メッセージ処理関数のタスク記録を直接確認できます。

サブスクライブするメッセージの数が膨大な場合、各メッセージの処理ログを記録すること自体がパフォーマンス問題を引き起こす可能性があります。デプロイとメンテナンス / システムメトリクスとタスク記録 / ローカル関数タスク記録の無効化を参照して「ローカル関数タスク記録」を無効にし、MySQL のストレージ負荷を軽減できます

旧版 DataFlux Func では、メッセージ処理関数はタスク記録の照会をサポートしていません。処理中に発生したエラーを記録したい場合は、メッセージ処理関数内で関連情報を DFF.CACHE に書き込むことができます

以下のコード実装を参照してください：

Python
import arrow
import traceback

def message_handler_impl(topic, message):
    # ここが実際のメッセージ処理関数
    # ここでは0除算エラーが発生すると仮定
    x = 100 / 0

@DFF.API('Message Handler')
def message_handler(topic, message):
    try:
        # 実際のメッセージ処理関数を呼び出す
        message_handler_impl(topic, message)

    except Exception as e:
        # 現在時刻を取得
        now_str = arrow.now('Asia/Shanghai').format('YYYY-MM-DD HH:mm:ss')

        # 完全なエラー情報を取得
        error_stack = traceback.format_exc()

        # エラー情報を DFF.CACHE に保存
        latest_error = '\n'.join([ '時間：', now_str, 'エラー情報：', error_stack ])
        DFF.CACHE.set('latest_error', latest_error)

        # エラーは再スローする必要がある
        raise

実行プロセスで問題が発生した場合は、「管理 / 関数キャッシュマネージャー」で詳細情報を確認できます

6. シングルサブスクライブとマルチサブスクライブ

バージョン 1.7.31 で追加

すべてのサブスクライバーは DataFlux Func の server サービスで実行されます。サブスクライバーがメッセージを受信すると、設定に基づいて関数実行タスクを生成し、worker に送信して実行します。

サブスクライバーは2つのサブスクライブ方法をサポートしています：

シングルサブスクライブ：server レプリカをいくつ起動しても、サブスクライバーは常にそのうちの1つの server でのみ実行されます
マルチサブスクライブ：複数の server レプリカを起動すると、サブスクライバーは各 server でそれぞれ実行されます

Redis サブスクライバーの場合、共有サブスクライブ処理をサポートしていないため、マルチサブスクライブではメッセージが重複して消費されるだけになります。したがって、現時点ではシングルサブスクライブのみがサポートされています。

MQTT、Kafka サブスクライバーでは、シングルサブスクライブまたはマルチサブスクライブを選択できます。

MQTT でマルチサブスクライブを使用する場合、Topic は $share/group_name/topic_name の形式で共有サブスクライブする必要があります。そうしないと、同じメッセージを重複して受信する原因になります

DataFlux Func に含まれるサービスや、複数の server レプリカを起動することについては、デプロイとメンテナンス / アーキテクチャ、スケーリングとリソース制限を参照してください

7. サブスクライブ処理速度の向上

バージョン 1.7.31 で追加

DataFlux Func では、サブスクライバーは server サービス内で実行されますが、関数は worker サービス内で実行されるため、サブスクライブ処理速度は、サブスクライブメッセージの受信速度とサブスクライブ関数の実行速度の2つの部分で構成されます。

サブスクライブするメッセージの数は多くないが、各メッセージの処理が複雑で時間がかかる場合は、worker の数を増やしてメッセージがタイムリーに処理されるようにする必要があります。

サブスクライブするメッセージの数が膨大な場合は、server と worker の数を同時に増やし、サブスクライバーを「マルチサブスクライブ」モードに設定する必要があります。

システムのスケーリング方法については、デプロイとメンテナンス / アーキテクチャ、スケーリングとリソース制限を参照してください

8. サブスクライブの制限

バージョン 1.7.31 で追加

サブスクライブするメッセージの数が膨大な場合、サーバー側のパブリッシュ-サブスクライブ方式が異なるため、対応する制限も異なります。

Redis / MQTT

Redis、MQTT サブスクライバーはサブスクライブ側でメッセージの受信速度を制御できないため、受信したメッセージは内部のバッファプールに入ります。

バッファプールのデフォルトサイズは 5000 です。つまり、最大 5000 件のサブスクライブメッセージが処理待ちで滞留できます。その間にさらに多くのメッセージを受信した場合、あふれたメッセージは破棄されます。

上記の問題を解決するには、worker の数を増やして、サブスクライブメッセージの処理速度を向上させることができます。

Kafka

Kafka サブスクライバーはサブスクライブ側でメッセージの受信速度を制御できるため、バッファプールは存在せず、メッセージも破棄されません。

メッセージ処理速度がメッセージのパブリッシュ速度に追いつかない場合は、worker の数を増やして、サブスクライブメッセージの処理速度を向上させることができます。

9. 既知の問題

現在の既知の問題は以下のとおりです。

MQTT の動作が遅く、メッセージを受信できない

MQTT Broker が EMQX に接続する際に、動作が遅くなったり、メッセージを受信できない問題が発生する可能性があります。

この問題は初期のコミュニティ版スタンドアロン EMQX で発生しましたが、他のバージョンの EMQX ではまだテストされていません。

ただし、mosquitto に接続した場合はこの問題は見られませんでした。

Kafka 起動直後は消費できない

Kafka コネクターはサブスクライブ後、最初の数分間は一時停止状態になることがあり、その後正常にメッセージの消費が行われます。

原因は Kafka の Rebalance メカニズムに関係している可能性があります