通过消费组消费数据

消费数据

基于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 {
       // 日志服务的服务接入点，请您根据实际情况填写。
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // 日志服务项目名称，请您根据实际情况填写。请从已创建项目中获取项目名称。
       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
   --------
   ......