Simple Log Service:Mengonsumsi log dengan kelompok konsumen

SDK

Kode berikut menunjukkan cara membuat consumer group:

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 {
         // Contoh ini mengambil ID AccessKey dan Rahasia AccessKey dari variabel lingkungan.
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // Masukkan nama proyek.
        String projectName = "ali-test-project";
        // Masukkan nama Logstore.
        String logstoreName = "ali-test-logstore";
        // Tetapkan endpoint untuk Simple Log Service. Contoh ini menggunakan endpoint wilayah China (Hangzhou). Ganti dengan endpoint yang sesuai.
        String host = "https://cn-hangzhou.log.aliyuncs.com";

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

        try {
            // Tetapkan nama 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;
        }
    }
}

Untuk contoh kode pengelolaan consumer group, lihat Use the Java SDK to manage consumer groups dan Use Simple Log Service SDK for Python to manage consumer groups.

API

Untuk membuat consumer group menggunakan API, lihat CreateConsumerGroup.

Untuk memverifikasi bahwa consumer group telah dibuat, lihat ListConsumerGroup.

CLI

Untuk membuat consumer group menggunakan CLI, lihat create_consumer_group.

Untuk memverifikasi bahwa consumer group telah dibuat, lihat list_consumer_group.

Contoh 1: Konsumsi menggunakan SDK

Contoh 2: Konsumsi menggunakan SDK dengan SPL

1. Tambahkan dependensi Maven.

   Dalam file pom.xml Anda, tambahkan dependensi berikut:

   <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. Buat kelas untuk menentukan logika pemrosesan log Anda.

   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;
       // Melacak waktu terakhir checkpoint disimpan.
       private long mLastSaveTime = 0;
       // Dipanggil sekali saat inisialisasi processor.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
       // Logika utama untuk mengonsumsi data. Semua exception harus ditangani dalam metode ini. Jangan melempar exception secara langsung.
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // Cetak data yang diambil.
           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();
           // Simpan checkpoint ke server setiap 30 detik. Jika worker berhenti secara tidak terduga, worker baru akan melanjutkan dari checkpoint terakhir, sehingga berpotensi memproses ulang sejumlah kecil data.
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // Parameter 'true' segera menyimpan checkpoint ke server. Secara default, checkpoint di memori secara otomatis disimpan ke server setiap 60 detik.
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // Parameter 'false' menyimpan checkpoint secara lokal. Checkpoint ini akan disimpan ke server oleh mekanisme pembaruan otomatis.
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
       // Metode ini dipanggil saat worker dimatikan. Anda dapat melakukan tugas pembersihan di sini.
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Simpan checkpoint ke server segera.
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

3. Buat factory untuk menghasilkan instance processor log Anda.

   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() {
           // Hasilkan instance consumer. Catatan: Setiap pemanggilan metode generatorProcessor harus mengembalikan objek SPLLogHubProcessor yang baru.
           return new SPLLogHubProcessor();
       }
   }

4. Buat kelas utama untuk mengonfigurasi dan menjalankan thread pekerja.

   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 {
       // Endpoint Simple Log Service. Ganti dengan endpoint Anda yang sebenarnya.
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // Nama proyek. Ganti dengan nama proyek yang sudah ada.
       private static String Project = "ali-test-project";
       // Nama Logstore. Ganti dengan nama Logstore yang sudah ada.
       private static String Logstore = "ali-test-logstore";
       // Anda tidak perlu membuat consumer group terlebih dahulu; program akan membuatnya secara otomatis saat pertama kali dijalankan.
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // Contoh ini mengambil ID AccessKey dan Rahasia AccessKey dari variabel lingkungan.
       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" adalah nama consumer, yang harus unik dalam satu consumer group. Untuk menyeimbangkan konsumsi di beberapa mesin, Anda dapat menggunakan pengenal unik seperti alamat IP mesin sebagai nama consumer.
           // maxFetchLogGroupSize menetapkan jumlah maksimum LogGroup yang diambil per permintaan. Nilai default biasanya sudah cukup. Anda dapat menyesuaikannya dengan menggunakan config.setMaxFetchLogGroupSize(100). Rentang nilai yang valid adalah (0, 1000].
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // Gunakan setQuery untuk menentukan pernyataan SPL guna memfilter log selama konsumsi.
           config.setQuery("* | where cast(body_bytes_sent as bigint) > 14000");
           ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), config);
           Thread thread = new Thread(worker);
           // Setelah thread dimulai, ClientWorker berjalan secara otomatis. ClientWorker mengimplementasikan antarmuka Runnable.
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // Panggil metode shutdown() worker untuk menghentikan instance consumer dan thread-nya.
           worker.shutdown();
           // ClientWorker membuat tugas asinkron. Setelah shutdown(), tunggu tugas-tugas tersebut selesai secara elegan. Disarankan untuk memberikan jeda 30 detik.
           Thread.sleep(30 * 1000);
       }
   }

5. Jalankan Main.java.

   Berikut adalah contoh output dari konsumsi log 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
   --------
   ......