Simple Log Service:Consumo de registros con un grupo de consumidores

SDK

El siguiente código crea un grupo de consumidores:

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 {
         // This example obtains the AccessKey ID and AccessKey secret from environment variables.
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // Enter the project name.
        String projectName = "ali-test-project";
        // Enter the Logstore name.
        String logstoreName = "ali-test-logstore";
        // Set the endpoint for Simple Log Service. This example uses the endpoint of the China (Hangzhou) region. Replace it 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 {
            // Set the consumer group name.
            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;
        }
    }
}

Para ver ejemplos de código sobre cómo gestionar grupos de consumidores, consulte Use the Java SDK to manage consumer groups y Use Simple Log Service SDK for Python to manage consumer groups.

API

Para crear un grupo de consumidores mediante la API, consulte CreateConsumerGroup.

Para comprobar que el grupo de consumidores se creó correctamente, consulte ListConsumerGroup.

CLI

Para crear un grupo de consumidores mediante la CLI, consulte create_consumer_group.

Para comprobar que el grupo de consumidores se creó correctamente, consulte list_consumer_group.

Ejemplo 1: consumo con SDK

Ejemplo 2: consumo con SDK y SPL

1. Agregue las dependencias de Maven.

   En su archivo pom.xml, agregue las siguientes dependencias:

   <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.50</version>
   </dependency>

2. Cree una clase para definir la lógica de procesamiento de sus registros.

   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;
       // Tracks the last time a checkpoint was saved.
       private long mLastSaveTime = 0;
       // Called once upon processor initialization.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
       // The main logic for consuming data. All exceptions must be handled within this method. Do not throw exceptions directly.
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // Print the fetched 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();
           // Save a checkpoint to the server every 30 seconds. If the worker terminates unexpectedly, the new worker resumes from the last checkpoint, potentially reprocessing a small amount of data.
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // A 'true' parameter flushes the checkpoint to the server immediately. By default, the in-memory checkpoint is automatically flushed to the server every 60 seconds.
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // A 'false' parameter caches the checkpoint locally. It will be flushed to the server by the auto-update mechanism.
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
       // This method is called when the worker is shutting down. You can perform cleanup tasks here.
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Save the checkpoint to the server immediately.
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

3. Cree una fábrica para generar instancias de su procesador de registros.

   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 instance. Note: Each call to the generatorProcessor method should return a new SPLLogHubProcessor object.
           return new SPLLogHubProcessor();
       }
   }

4. Cree una clase principal para configurar e iniciar el hilo de trabajo.

   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 endpoint of Simple Log Service. Replace with your actual endpoint.
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // The project name. Replace with the name of an existing project.
       private static String Project = "ali-test-project";
       // The Logstore name. Replace with the name of an existing Logstore.
       private static String Logstore = "ali-test-logstore";
       // You do not need to create the consumer group beforehand; the program automatically creates it on the first run.
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // This example retrieves the AccessKey ID and AccessKey Secret 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" is the consumer name, which must be unique within a consumer group. To balance consumption across multiple machines, you can use unique identifiers like the machine's IP address as the consumer name.
           // maxFetchLogGroupSize sets the maximum number of LogGroups to fetch per request. The default value is usually sufficient. You can adjust it by using config.setMaxFetchLogGroupSize(100). The valid range is (0, 1000].
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // Use setQuery to specify an SPL statement to filter logs during 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 starts, the ClientWorker runs automatically. ClientWorker implements the Runnable interface.
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // Call the worker's shutdown() method to stop the consumer instance and its associated thread.
           worker.shutdown();
           // The ClientWorker creates asynchronous tasks. After shutdown(), wait for these tasks to complete gracefully. A 30-second sleep is recommended.
           Thread.sleep(30 * 1000);
       }
   }

5. Ejecute Main.java.

   A continuación se muestra un ejemplo de la salida al consumir registros de 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
   --------
   ......

SDK de Java

1. Consulte el checkpoint de consumo de cada shard. El siguiente código muestra un ejemplo:

   ConsumerGroupTest.java

   import java.util.List;
   import com.aliyun.openservices.log.Client;
   import com.aliyun.openservices.log.common.Consts.CursorMode;
   import com.aliyun.openservices.log.common.ConsumerGroup;
   import com.aliyun.openservices.log.common.ConsumerGroupShardCheckPoint;
   import com.aliyun.openservices.log.exception.LogException;
   public class ConsumerGroupTest {
       static String endpoint = "cn-hangzhou.log.aliyuncs.com";
       static String project = "ali-test-project";
       static String logstore = "ali-test-logstore";
       static String accesskeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
       static String accesskey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
       public static void main(String[] args) throws LogException {
           Client client = new Client(endpoint, accesskeyId, accesskey);
           // Get all consumer groups in the Logstore. The list is empty if no consumer groups exist.
           List<ConsumerGroup> consumerGroups = client.ListConsumerGroup(project, logstore).GetConsumerGroups();
           for(ConsumerGroup c: consumerGroups){
               // Print the consumer group properties: name, heartbeat timeout, and ordered consumption status.
               System.out.println("Name: " + c.getConsumerGroupName());
               System.out.println("heartbeat timeout: " + c.getTimeout());
               System.out.println("ordered consumption: " + c.isInOrder());
               for(ConsumerGroupShardCheckPoint cp: client.GetCheckPoint(project, logstore, c.getConsumerGroupName()).GetCheckPoints()){
                   System.out.println("shard: " + cp.getShard());
                   // The time is a long integer, accurate to the microsecond.
                   System.out.println("Last checkpoint update time: " + cp.getUpdateTime());
                   System.out.println("Consumer name: " + cp.getConsumer());
                   String consumerPrg = "";
                   if(cp.getCheckPoint().isEmpty())
                       consumerPrg = "Consumption has not started";
                   else{
                       // A UNIX timestamp in seconds. Format the output as needed.
                       try{
                           int prg = client.GetPrevCursorTime(project, logstore, cp.getShard(), cp.getCheckPoint()).GetCursorTime();
                           consumerPrg = "" + prg;
                       }
                       catch(LogException e){
                           if(e.GetErrorCode() == "InvalidCursor")
                               consumerPrg = "Invalid. The consumption checkpoint is older than the data retention period.";
                           else{
                               // internal server error
                               throw e;
                           }
                       }
                   }
                   System.out.println("consumption checkpoint: " + consumerPrg);
                   String endCursor = client.GetCursor(project, logstore, cp.getShard(), CursorMode.END).GetCursor();
                   int endPrg = 0;
                   try{
                       endPrg = client.GetPrevCursorTime(project, logstore, cp.getShard(), endCursor).GetCursorTime();
                   }
                   catch(LogException e){
                       // do nothing
                   }
                   // A UNIX timestamp in seconds. Format the output as needed.
                   System.out.println("Arrival time of the last record: " + endPrg);
               }
           }
       }
   }

2. A continuación se muestra un ejemplo de la salida:

   Name: ali-test-consumergroup2
   heartbeat timeout: 60
   ordered consumption: false
   shard: 0
   Last checkpoint update time: 0
   Consumer name: consumer_1
   consumption checkpoint: Consumption has not started
   Arrival time of the last record: 1729583617
   shard: 1
   Last checkpoint update time: 0
   Consumer name: consumer_1
   consumption checkpoint: Consumption has not started
   Arrival time of the last record: 1729583738
   Process finished with exit code 0

Consola

1. Inicie sesión en la consola de Simple Log Service.

2. En la sección Projects, haga clic en el proyecto deseado.

   

3. En la pestaña Log Storage > Logstores, haga clic en el icono situado a la izquierda del logstore correspondiente y, a continuación, haga clic en el icono a la izquierda de Data Consumption.

4. En la lista de grupos de consumidores, haga clic en el grupo deseado.

5. En la página Consumer Group Status, consulte el checkpoint de consumo de cada shard. Esta página muestra los detalles de cada shard, incluidos su identificador (shard), la hora del último consumo (Last Consumed Time) y el Client que lo consume. También puede usar los botones Refresh y Reset Checkpoint.