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 {
         // In this example, the AccessKey ID and AccessKey secret are obtained from environment variables. 
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // The name of the project. 
        String projectName = "ali-test-project";
        // The name of the Logstore. 
        String logstoreName = "ali-test-logstore";
        // The Simple Log Service endpoint. In this example, the Simple Log Service endpoint for the China (Hangzhou) region is used. Replace the parameter value with the actual endpoint. 
        String host = "https://cn-hangzhou.log.aliyuncs.com";

        // Create a Simple Log Service client. 
        Client client = new Client(host, accessId, accessKey);

        try {
            // The name of the consumer group. 
            String consumerGroupName = "ali-test-consumergroup2";
            System.out.println("ready to create consumergroup");

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

            client.CreateConsumerGroup(projectName, logstoreName, consumerGroup);

            System.out.println(String.format("create consumergroup %s success", consumerGroupName));

        } catch (LogException e) {
            System.out.println("LogException e :" + e.toString());
            System.out.println("error code :" + e.GetErrorCode());
            System.out.println("error message :" + e.GetErrorMessage());
            throw e;
        }
    }
}

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

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;
       // The time at which the last consumption 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 consumption 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 consumption checkpoint. A small amount of data may be repeatedly consumed. 
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // The value true indicates that the consumption checkpoints are immediately updated to Simple Log Service. By default, the consumption 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 consumption checkpoints are locally cached and can be updated to Simple Log Service by using the automatic consumption 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 consumption checkpoints. 
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Immediately store consumption checkpoints to 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() {
           // Generate a consumer. Each time the generatorProcessor method is called, a new SPLLogHubProcessor object is returned as expected. 
           return new SPLLogHubProcessor();
       }
   }

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

   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 {
       // 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-test-project";
       // 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 = "ali-test-logstore";
       // 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 = "ali-test-consumergroup2";
       // 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);
           // You can use setQuery to specify a Simple Log Service Processing Language (SPL) statement for data consumption.
           config.setQuery("* | where cast(body_bytes_sent as bigint) > 14000");
           ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), 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
   --------
   ......