Simple Log Service：通過消費組消費資料

消費資料

基於SPL消費資料

您可以通過Java、C++、Python及Go SDK實現消費組消費資料。此處，以Java SDK為例。

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>aliyun-log</artifactId>
         <version>0.6.99</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;
       // 記錄上次持久化Checkpoint的時間。
       private long mLastSaveTime = 0;
   
       // initialize 方法會在 processor 對象初始化時被調用一次
       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("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();
           // 每隔30秒，寫一次Checkpoint到服務端。如果30秒內發生Worker異常終止，新啟動的Worker會從上一個Checkpoint擷取消費資料，可能存在少量的重複資料。
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // 參數為true表示立即手動將Checkpoint更新到服務端。此外，預設每60秒會自動將記憶體中緩衝的Checkpoint更新到服務端。
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // 參數為false表示將Checkpoint緩衝在本地，可被自動更新Checkpoint機制更新到服務端。
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
   
       // 當Worker退出時，會調用該函數，您可以在此處執行清理工作。
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // 將Checkpoint立即儲存到服務端。
           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檔案。建立一個消費組並啟動一個消費者線程，該消費者會從指定的Logstore中消費資料。管理消費組的程式碼範例，請參見使用Java SDK管理消費組、使用Python SDK管理消費組。

   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 SPLConsumer {
       // Log Service的服務存取點，請您根據實際情況填寫。
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // Log Service專案名稱，請您根據實際情況填寫。請從已建立專案中擷取專案名稱。
       private static String Project = "your_project";
       // 日誌庫名稱，請您根據實際情況填寫。請從已建立日誌庫中擷取日誌庫名稱。
       private static String Logstore = "your_logstore";
       // 消費組名稱，請您根據實際情況填寫。您無需提前建立，該程式運行時會自動建立該消費組。
       private static String ConsumerGroup = "consumerGroupX";
       // 本樣本從環境變數中擷取AccessKey ID和AccessKey Secret。。
       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是消費者名稱，同一個消費組下面的消費者名稱必須不同。不同消費者在多台機器上啟動多個進程，均衡消費一個Logstore時，消費者名稱可以使用機器IP地址來區分。
           // maxFetchLogGroupSize用於設定每次從服務端擷取的LogGroup最大數目，使用預設值即可。您可以使用config.setMaxFetchLogGroupSize(100);調整，取值範圍為(0,1000]。
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // setQuery可以設定消費過程中的SLS 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會自動運行，ClientWorker擴充了Runnable介面。
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // 調用Worker的Shutdown函數，退出消費執行個體，關聯的線程也會自動停止。
           worker.shutdown();
           // ClientWorker運行過程中會產生多個非同步任務。Shutdown完成後，請等待還在執行的任務安全退出。建議設定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
   --------
   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
   --------
   ......