Simple Log Service:Use consumer groups to consume data

You can use Simple Log Service SDK for Java, C++, Python, or Go to create consumer groups and consume data. In this example, Simple Log Service SDK for Java is used.

1. Add Maven dependencies.

   Open the pom.xml file in the root directory of your Java project and add the following code:

   <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. Create a file named SampleLogHubProcessor.java to write the implementation logic of consumers.

   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();
           }
       }
   }

   For more information, see Java, C++, Python, and Go.

3. Create a file named SampleLogHubProcessorFactory.java to define a consumer entity.

   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. Create a Main.java file. Create a consumer group and start a consumer thread to allow consumers in the consumer group to consume data in the specified 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. Run the Main.java file.

   In this example, NGINX logs are consumed and the consumption result is displayed.

   :    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
   --------
   ......

How it works

The first time you call Simple Log Service SDK for Java to start a consumer, the SDK creates a consumer group if the SDK does not find the consumer group to which the consumer belongs. After the consumer group is created, the SDK records a start checkpoint and starts to consume data from the checkpoint. The start checkpoint becomes invalid after the first-time consumption. When the consumer is restarted, the consumer resumes data consumption from the last checkpoint that is stored by Simple Log Service. Sample checkpoints:

• LogHubConfig.ConsumePosition.BEGIN_CURSOR: the start checkpoint, which specifies the first log in a Logstore. A consumer starts consumption from the earliest data.

• LogHubConfig.ConsumePosition.END_CURSOR: the end checkpoint, which specifies the last log in a Logstore.