Simple Log Service：Gunakan grup konsumen untuk mengonsumsi data

Anda dapat menggunakan SDK Layanan Log Sederhana untuk Java, C++, Python, atau Go untuk membuat grup konsumen dan mengonsumsi data. Dalam contoh ini, SDK Layanan Log Sederhana untuk Java digunakan.

1. Tambahkan dependensi Maven.

   Buka file pom.xml di direktori root proyek Java Anda dan tambahkan kode 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.47</version>
   </dependency>

2. Buat file bernama SampleLogHubProcessor.java untuk menulis logika implementasi konsumen.

   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;
       // Waktu terakhir titik pemeriksaan disimpan.
       private long mLastSaveTime = 0;
   
       // Metode initialize dipanggil sekali saat objek prosesor diinisialisasi.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
   
       // Logika utama konsumsi data. Anda harus menyertakan kode untuk menangani semua pengecualian yang mungkin terjadi selama konsumsi data.
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // Tampilkan data yang diperoleh.
           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();
           // Titik pemeriksaan ditulis ke Layanan Log Sederhana pada interval 30 detik. Jika instance ClientWorker berhenti secara tak terduga dalam 30 detik, instance ClientWorker yang baru dimulai mengonsumsi data dari titik pemeriksaan terakhir. Sejumlah kecil data mungkin dikonsumsi berulang kali.
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // Nilai true menunjukkan bahwa titik pemeriksaan segera diperbarui ke Layanan Log Sederhana. Secara default, titik pemeriksaan yang disimpan di memori secara otomatis diperbarui ke Layanan Log Sederhana pada interval 60 detik.
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // Nilai false menunjukkan bahwa titik pemeriksaan disimpan secara lokal dan dapat diperbarui ke Layanan Log Sederhana menggunakan mekanisme pembaruan titik pemeriksaan otomatis.
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
   
       // Fungsi shutdown instance ClientWorker dipanggil. Anda dapat mengelola titik pemeriksaan.
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Segera simpan titik pemeriksaan ke Layanan Log Sederhana.
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

   Untuk informasi lebih lanjut, lihat Java, C++, Python, dan Go.

3. Buat file bernama SampleLogHubProcessorFactory.java untuk mendefinisikan entitas konsumen.

   class SampleLogHubProcessorFactory implements ILogHubProcessorFactory {
       public ILogHubProcessor generatorProcessor() {
           // Hasilkan konsumen. Setiap kali metode generatorProcessor dipanggil, objek SampleLogHubProcessor baru dikembalikan sesuai harapan.
           return new SampleLogHubProcessor();
       }
   }

4. Buat file Main.java. Buat grup konsumen dan mulai thread konsumen untuk memungkinkan konsumen dalam grup konsumen mengonsumsi data di penyimpanan log yang ditentukan.

   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 {
       // Titik akhir Layanan Log Sederhana. Masukkan titik akhir berdasarkan kebutuhan bisnis Anda.
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // Nama proyek Layanan Log Sederhana. Masukkan nama berdasarkan kebutuhan bisnis Anda. Anda harus memasukkan nama proyek yang ada.
       private static String Project = "ali-cn-hangzhou-sls-admin";
       // Nama penyimpanan log. Masukkan nama berdasarkan kebutuhan bisnis Anda. Anda harus memasukkan nama penyimpanan log yang ada.
       private static String Logstore = "sls_operation_log";
       // Nama grup konsumen. Masukkan nama berdasarkan kebutuhan bisnis Anda. Anda tidak perlu membuat grup konsumen terlebih dahulu. Grup konsumen dibuat secara otomatis saat program berjalan.
       private static String ConsumerGroup = "consumerGroupX";
       // Dalam contoh ini, ID AccessKey dan rahasia AccessKey diperoleh 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 menentukan nama konsumen. Nama setiap konsumen dalam grup konsumen harus unik. Jika konsumen berbeda memulai proses pada mesin berbeda untuk mengonsumsi data dalam penyimpanan log, Anda dapat menggunakan alamat IP mesin untuk mengidentifikasi setiap konsumen.
           // maxFetchLogGroupSize menentukan jumlah maksimum grup log yang dapat diperoleh dari Layanan Log Sederhana sekaligus. Pertahankan nilai default. Anda dapat menggunakan config.setMaxFetchLogGroupSize(100); untuk mengubah jumlah maksimum. Nilai valid: (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);
           // Setelah instance Thread berjalan, instance ClientWorker secara otomatis berjalan dan memperluas antarmuka Runnable.
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // Fungsi shutdown instance ClientWorker dipanggil untuk keluar dari konsumen. Instance Thread secara otomatis berhenti.
           worker.shutdown();
           // Beberapa tugas asinkron dihasilkan saat instance ClientWorker berjalan. Untuk memastikan bahwa semua tugas yang sedang berjalan berhenti dengan aman setelah shutdown, kami sarankan Anda mengatur Thread.sleep selama 30 detik.
           Thread.sleep(30 * 1000);
       }
   }

5. Jalankan file Main.java.

   Dalam contoh ini, log NGINX dikonsumsi dan hasil konsumsi ditampilkan.

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

Cara kerjanya

Pertama kali Anda memanggil SDK Layanan Log Sederhana untuk Java untuk memulai konsumen, SDK akan membuat grup konsumen jika tidak menemukan grup konsumen tempat konsumen tersebut milik. Setelah grup konsumen dibuat, SDK mencatat titik pemeriksaan awal dan mulai mengonsumsi data dari titik pemeriksaan tersebut. Titik pemeriksaan awal menjadi tidak valid setelah konsumsi pertama kali. Saat konsumen dimulai ulang, konsumen melanjutkan konsumsi data dari titik pemeriksaan terakhir yang disimpan oleh Layanan Log Sederhana. Contoh titik pemeriksaan:

• LogHubConfig.ConsumePosition.BEGIN_CURSOR: titik pemeriksaan awal, yang menentukan log pertama dalam penyimpanan log. Seorang konsumen memulai konsumsi dari data paling awal.

• LogHubConfig.ConsumePosition.END_CURSOR: titik pemeriksaan akhir, yang menentukan log terakhir dalam penyimpanan log.