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

Simple Log Service SDK for Java、C ++ 、Python、またはGoを使用して、コンシューマーグループを作成し、データを消費できます。 この例では、Simple Log Service SDK for Javaが使用されています。

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

   Javaプロジェクトのルートディレクトリにある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>loghub-client-lib</artifactId>
     <version>0.6.47</version>
   </dependency>

2. コンシューマの実装ロジックを記述するには、SampleLogHubProcessor.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 SampleLogHubProcessor implements ILogHubProcessor {
       private int shardId;
       // The time at which the last checkpoint was stored. 
       private long mLastSaveTime = 0;
   
       // The initialize method is called once when a processor object is initialized.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
   
       // The main logic of data consumption. You must include the code to handle all exceptions that may occur during data consumption. 
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // Display the obtained data. 
           for (LogGroupData logGroup : logGroups) {
               FastLogGroup fastLogGroup = logGroup.GetFastLogGroup();
               System.out.println("Tags");
               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("--------\nLog: " + i + ", time: " + 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();
           // A checkpoint is written to Simple Log Service at an interval of 30 seconds. If the ClientWorker instance unexpectedly stops within 30 seconds, a newly started ClientWorker instance consumes data from the last checkpoint. A small amount of data may be repeatedly consumed. 
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // The value true indicates that the checkpoints are immediately updated to Simple Log Service. By default, the checkpoints that are cached in memory are automatically updated to Simple Log Service at an interval of 60 seconds. 
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // The value false indicates that the checkpoints are locally cached and can be updated to Simple Log Service by using the automatic checkpoint update mechanism. 
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
   
       // The shutdown function of the ClientWorker instance is called. You can manage the checkpoints. 
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Immediately store checkpoints to Simple Log Service. 
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

   詳細については、「Java」、「C ++ 」、「Python」、および「Go」をご参照ください。

3. コンシューマエンティティを定義するために、SampleLogHubProcessorFactory.javaという名前のファイルを作成します。

   class SampleLogHubProcessorFactory implements ILogHubProcessorFactory {
       public ILogHubProcessor generatorProcessor() {
           // Generate a consumer. Each time the generatorProcessor method is called, a new SampleLogHubProcessor object is returned as expected. 
           return new SampleLogHubProcessor();
       }
   }

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

   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 {
       // The Simple Log Service endpoint. Enter an endpoint based on your business requirements. 
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // The name of the Simple Log Service project. Enter a name based on your business requirements. You must enter the name of an existing project. 
       private static String Project = "ali-cn-hangzhou-sls-admin";
       // The name of the Logstore. Enter a name based on your business requirements. You must enter the name of an existing Logstore. 
       private static String Logstore = "sls_operation_log";
       // The name of the consumer group. Enter a name based on your business requirements. You do not need to create a consumer group in advance. A consumer group is automatically created when a program runs. 
       private static String ConsumerGroup = "consumerGroupX";
       // In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.  
       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 specifies the name of a consumer. The name of each consumer in a consumer group must be unique. If different consumers start processes on different machines to consume data in a Logstore, you can use the machine IP addresses to identify each consumer. 
           // maxFetchLogGroupSize specifies the maximum number of log groups that can be obtained from Simple Log Service at a time. Retain the default value. You can use config.setMaxFetchLogGroupSize(100); to change the maximum number. Valid values: (0,1000]. 
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR,1000);
           ClientWorker worker = new ClientWorker(new SampleLogHubProcessorFactory(), config);
           Thread thread = new Thread(worker);
           // After the Thread instance runs, the ClientWorker instance automatically runs and extends the Runnable interface. 
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // The shutdown function of the ClientWorker instance is called to exit the consumer. The Thread instance is automatically stopped. 
           worker.shutdown();
           // Multiple asynchronous tasks are generated when the ClientWorker instance is running. To ensure that all running tasks securely stop after the shutdown, we recommend that you set Thread.sleep to 30 seconds. 
           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
   --------
   Log: 158, time: 1635629778, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   Tags
       __receive_time__    :    1635629815
   --------
   Log: 0, time: 1635629788, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   Tags
       __receive_time__    :    1635629877
   --------
   ......

仕組み

Simple Log Service SDK for Javaを初めて呼び出してコンシューマーを起動すると、コンシューマーが属するコンシューマーグループが見つからない場合、SDKはコンシューマーグループを作成します。 コンシューマーグループが作成されると、SDKは開始チェックポイントを記録し、チェックポイントからデータの消費を開始します。 開始チェックポイントは、初回消費後に無効になります。 コンシューマーが再起動されると、コンシューマーはSimple Log Serviceによって保存された最後のチェックポイントからデータの消費を再開します。 サンプルチェックポイント:

• LogHubConfig.ConsumePosition.BEGIN_CURSOR: 開始チェックポイント。Logstoreの最初のログを指定します。 消費者は、最も早いデータから消費を開始する。

• LogHubConfig.ConsumePosition.END_CURSOR: ログストア内の最後のログを指定する終了チェックポイント。