Simple Log Service:コンシューマーグループを使用したログの消費

Simple Log Service SDK を使用する

サンプルコード：

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.ConsumerGroup;
import com.aliyun.openservices.log.exception.LogException;

public class CreateConsumerGroup {
    public static void main(String[] args) throws LogException {
         // この例では、アクセスキー ID とアクセスキーシークレットは環境変数から取得されます。
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // プロジェクトの名前。
        String projectName = "ali-test-project";
        // ログストアの名前。
        String logstoreName = "ali-test-logstore";
        // Simple Log Service エンドポイント。この例では、中国 (杭州) リージョンの Simple Log Service エンドポイントが使用されています。パラメータ値を実際のエンドポイントに置き換えてください。
        String host = "https://cn-hangzhou.log.aliyuncs.com";

        // Simple Log Service クライアントを作成します。
        Client client = new Client(host, accessId, accessKey);

        try {
            // コンシューマーグループの名前。
            String consumerGroupName = "ali-test-consumergroup2";
            System.out.println("コンシューマーグループを作成する準備ができました");

            ConsumerGroup consumerGroup = new ConsumerGroup(consumerGroupName, 300, true);

            client.CreateConsumerGroup(projectName, logstoreName, consumerGroup);

            System.out.println(String.format("コンシューマーグループ %s の作成に成功しました", consumerGroupName));

        } catch (LogException e) {
            System.out.println("LogException e :" + e.toString());
            System.out.println("エラーコード :" + e.GetErrorCode());
            System.out.println("エラーメッセージ :" + e.GetErrorMessage());
            throw e;
        }
    }
}

コンシューマーグループの管理に使用されるサンプルコードの詳細については、「Java 用 Simple Log Service SDK を使用してコンシューマーグループを管理する」および「Python 用 Simple Log Service SDK を使用してコンシューマーグループを管理する」をご参照ください。

Simple Log Service API を使用する

Simple Log Service API を使用してコンシューマーグループを作成する方法の詳細については、「CreateConsumerGroup」をご参照ください。

コンシューマーグループが作成されているかどうかを確認する方法の詳細については、「ListConsumerGroup」をご参照ください。

Simple Log Service CLI を使用する

Simple Log Service CLI を使用してコンシューマーグループを作成する方法の詳細については、「create_consumer_group」をご参照ください。

コンシューマーグループが作成されているかどうかを確認する方法の詳細については、「list_consumer_group」をご参照ください。

例 1：SDK を使用する

例 2：SDK と SPL を使用する

1. Maven の依存関係を追加します。

   pom.xml ファイルを開き、次のコードを追加します。

       <dependency>
         <groupId>com.google.protobuf</groupId>
         <artifactId>protobuf-java</artifactId>
         <version>2.5.0</version>
       </dependency>
       <dependency>
         <groupId>com.aliyun.openservices</groupId>
         <artifactId>aliyun-log</artifactId>
         <version>0.6.114</version>
       </dependency>

2. データ消費の実装ロジックを記述します。サンプルコード：

   SPLLogHubProcessor.java

   import com.aliyun.openservices.log.common.FastLog;
   import com.aliyun.openservices.log.common.FastLogContent;
   import com.aliyun.openservices.log.common.FastLogGroup;
   import com.aliyun.openservices.log.common.FastLogTag;
   import com.aliyun.openservices.log.common.LogGroupData;
   import com.aliyun.openservices.loghub.client.ILogHubCheckPointTracker;
   import com.aliyun.openservices.loghub.client.exceptions.LogHubCheckPointException;
   import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;
   
   import java.util.List;
   
   public class SPLLogHubProcessor implements ILogHubProcessor {
       private int shardId;
       // 最後の消費チェックポイントが保存された時刻。
       private long mLastSaveTime = 0;
   
       // プロセッサオブジェクトが初期化されるときに、initialize メソッドが 1 回呼び出されます。
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
   
       // データ消費のメインロジック。データ消費中に発生する可能性のあるすべての例外を処理するコードを含める必要があります。
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // 取得したデータを表示します。
           for (LogGroupData logGroup : logGroups) {
               FastLogGroup fastLogGroup = logGroup.GetFastLogGroup();
               System.out.println("タグ");
               for (int i = 0; i < fastLogGroup.getLogTagsCount(); ++i) {
                   FastLogTag logTag = fastLogGroup.getLogTags(i);
                   System.out.printf("%s : %s\n", logTag.getKey(), logTag.getValue());
               }
               for (int i = 0; i < fastLogGroup.getLogsCount(); ++i) {
                   FastLog log = fastLogGroup.getLogs(i);
                   System.out.println("--------\nログ: " + i + ", 時刻: " + log.getTime() + ", GetContentCount: " + log.getContentsCount());
                   for (int j = 0; j < log.getContentsCount(); ++j) {
                       FastLogContent content = log.getContents(j);
                       System.out.println(content.getKey() + "\t:\t" + content.getValue());
                   }
               }
           }
           long curTime = System.currentTimeMillis();
           // 消費チェックポイントは 30 秒間隔で Simple Log Service に書き込まれます。ClientWorker インスタンスが 30 秒以内に予期せず停止した場合、新しく起動された ClientWorker インスタンスは最後の消費チェックポイントからデータを消費します。少量のデータが繰り返し消費される可能性があります。
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // 値 true は、消費チェックポイントが Simple Log Service にすぐに更新されることを示します。デフォルトでは、メモリにキャッシュされた消費チェックポイントは 60 秒間隔で Simple Log Service に自動的に更新されます。
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // 値 false は、消費チェックポイントがローカルにキャッシュされ、自動消費チェックポイント更新メカニズムを使用して Simple Log Service に更新できることを示します。
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
   
       // ClientWorker インスタンスのシャットダウン関数が呼び出されます。消費チェックポイントを管理できます。
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // 消費チェックポイントを Simple Log Service にすぐに保存します。
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

3. コンシューマーエンティティを定義します。サンプルコード：

   SPLLogHubProcessorFactory.java

   import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;
   import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessorFactory;
   
   class SPLLogHubProcessorFactory implements ILogHubProcessorFactory {
       public ILogHubProcessor generatorProcessor() {
           // コンシューマーを生成します。generatorProcessor メソッドが呼び出されるたびに、新しい SPLLogHubProcessor オブジェクトが期待どおりに返されます。
           return new SPLLogHubProcessor();
       }
   }

4. コンシューマーグループを作成し、コンシューマースレッドを起動して、コンシューマーグループ内のコンシューマーが指定されたログストア内のデータを消費できるようにします。サンプルコード：

   Main.java

   import com.aliyun.openservices.loghub.client.ClientWorker;
   import com.aliyun.openservices.loghub.client.config.LogHubConfig;
   import com.aliyun.openservices.loghub.client.exceptions.LogHubClientWorkerException;
   
   public class Main {
       // Simple Log Service エンドポイント。ビジネス要件に基づいてエンドポイントを入力します。
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // Simple Log Service プロジェクトの名前。ビジネス要件に基づいて名前を入力します。既存のプロジェクトの名前を入力する必要があります。
       private static String Project = "ali-test-project";
       // ログストアの名前。ビジネス要件に基づいて名前を入力します。既存のログストアの名前を入力する必要があります。
       private static String Logstore = "ali-test-logstore";
       // コンシューマーグループの名前。ビジネス要件に基づいて名前を入力します。事前にコンシューマーグループを作成する必要はありません。プログラムの実行時にコンシューマーグループが自動的に作成されます。
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // この例では、アクセスキー ID とアクセスキーシークレットは環境変数から取得されます。
       private static String AccessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
       private static String AccessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
   
   
       public static void main(String[] args) throws LogHubClientWorkerException, InterruptedException {
           // consumer_1 はコンシューマーの名前を指定します。コンシューマーグループ内の各コンシューマーの名前は一意である必要があります。異なるコンシューマーが異なるマシンでプロセスを起動してログストア内のデータを消費する場合、マシンの IP アドレスを使用して各コンシューマーを識別できます。
           // maxFetchLogGroupSize は、Simple Log Service から一度に取得できるロググループの最大数を指定します。デフォルト値を保持します。config.setMaxFetchLogGroupSize(100); を使用して最大数を変更できます。有効な値：(0,1000]。
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // setQuery を使用して、データ消費用の Simple Log Service 処理言語 (SPL) ステートメントを指定できます。
           config.setQuery("* | where cast(body_bytes_sent as bigint) > 14000");
           ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), config);
           Thread thread = new Thread(worker);
           // Thread インスタンスが実行されると、ClientWorker インスタンスが自動的に実行され、Runnable インターフェースが拡張されます。
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // ClientWorker インスタンスの shutdown 関数が呼び出され、コンシューマーが終了します。Thread インスタンスは自動的に停止します。
           worker.shutdown();
           // ClientWorker インスタンスの実行中に複数の非同期タスクが生成されます。shutdown 後にすべての実行中タスクが安全に停止するように、Thread.sleep を 30 秒に設定することをお勧めします。
           Thread.sleep(30 * 1000);
       }
   }

5. Main.java ファイルを実行します。

   この例では、NGINX ログが消費され、消費結果が表示されます。

   :    GET
   request_uri    :    /request/path-3/file-7
   status    :    200
   body_bytes_sent    :    3820
   host    :    www.example.com
   request_time    :    43
   request_length    :    1987
   http_user_agent    :    Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36
   http_referer    :    www.example.com
   http_x_forwarded_for    :    192.168.10.196
   upstream_response_time    :    0.02
   --------
   ログ: 158, 時刻: 1635629778, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   タグ
       __receive_time__    :    1635629815
   --------
   ログ: 0, 時刻: 1635629788, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   タグ
       __receive_time__    :    1635629877
   --------
   ......