Simple Log Service：Gunakan kelompok konsumen untuk mengonsumsi log

Gunakan SDK Layanan Log Sederhana

Contoh kode:

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 {
         // Dalam contoh ini, ID AccessKey dan rahasia AccessKey diperoleh dari variabel lingkungan.
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // Nama proyek.
        String projectName = "ali-test-project";
        // Nama penyimpanan log.
        String logstoreName = "ali-test-logstore";
        // Titik akhir Layanan Log Sederhana. Dalam contoh ini, titik akhir Layanan Log Sederhana untuk wilayah China (Hangzhou) digunakan. Ganti nilai parameter dengan titik akhir yang sebenarnya.
        String host = "https://cn-hangzhou.log.aliyuncs.com";

        // Buat klien Layanan Log Sederhana.
        Client client = new Client(host, accessId, accessKey);

        try {
            // Nama kelompok konsumen.
            String consumerGroupName = "ali-test-consumergroup2";
            System.out.println("siap untuk membuat consumergroup");

            ConsumerGroup consumerGroup = new ConsumerGroup(consumerGroupName, 300, true);

            client.CreateConsumerGroup(projectName, logstoreName, consumerGroup);

            System.out.println(String.format("buat consumergroup %s berhasil", consumerGroupName));

        } catch (LogException e) {
            System.out.println("LogException e :" + e.toString());
            System.out.println("kode kesalahan :" + e.GetErrorCode());
            System.out.println("pesan kesalahan :" + e.GetErrorMessage());
            throw e;
        }
    }
}

Untuk informasi lebih lanjut tentang contoh kode yang digunakan untuk mengelola kelompok konsumen, lihat Gunakan SDK Layanan Log Sederhana untuk Java untuk mengelola kelompok konsumen dan Gunakan SDK Layanan Log Sederhana untuk Python untuk mengelola kelompok konsumen.

Gunakan API Layanan Log Sederhana

Untuk informasi lebih lanjut tentang cara menggunakan API Layanan Log Sederhana untuk membuat kelompok konsumen, lihat CreateConsumerGroup.

Untuk informasi lebih lanjut tentang cara memeriksa apakah kelompok konsumen telah dibuat, lihat ListConsumerGroup.

Gunakan CLI Layanan Log Sederhana

Untuk informasi lebih lanjut tentang cara menggunakan CLI Layanan Log Sederhana untuk membuat kelompok konsumen, lihat create_consumer_group.

Untuk informasi lebih lanjut tentang cara memeriksa apakah kelompok konsumen telah dibuat, lihat list_consumer_group.

Contoh 1: Gunakan SDK

Contoh 2: Gunakan SDK dan SPL

1. Tambahkan dependensi Maven.

   Buka file pom.xml 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. Tulis logika implementasi konsumsi data. Contoh kode:

   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;
       // Waktu checkpoint konsumsi terakhir 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 + ", waktu: " + 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();
           // Checkpoint konsumsi 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 checkpoint konsumsi terakhir. Sejumlah kecil data mungkin dikonsumsi berulang kali.
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // Nilai true menunjukkan bahwa checkpoint konsumsi segera diperbarui ke Layanan Log Sederhana. Secara default, checkpoint konsumsi 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 checkpoint konsumsi disimpan secara lokal dan dapat diperbarui ke Layanan Log Sederhana menggunakan mekanisme pembaruan checkpoint konsumsi otomatis.
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
   
       // Fungsi shutdown dari instance ClientWorker dipanggil. Anda dapat mengelola checkpoint konsumsi.
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Segera simpan checkpoint konsumsi ke Layanan Log Sederhana.
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

3. Definisikan entitas konsumen. Contoh kode:

   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 konsumen. Setiap kali metode generatorProcessor dipanggil, objek SPLLogHubProcessor baru dikembalikan sesuai harapan.
           return new SPLLogHubProcessor();
       }
   }

4. Buat kelompok konsumen dan mulai thread konsumen untuk memungkinkan konsumen dalam kelompok konsumen mengonsumsi data dalam penyimpanan log yang ditentukan. Contoh kode:

   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 {
       // 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-test-project";
       // Nama penyimpanan log. Masukkan nama berdasarkan kebutuhan bisnis Anda. Anda harus memasukkan nama penyimpanan log yang ada.
       private static String Logstore = "ali-test-logstore";
       // Nama kelompok konsumen. Masukkan nama berdasarkan kebutuhan bisnis Anda. Anda tidak perlu membuat kelompok konsumen terlebih dahulu. Kelompok konsumen dibuat secara otomatis saat program berjalan.
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // 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 kelompok 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);
           // Anda dapat menggunakan setQuery untuk menentukan pernyataan Bahasa Pemrosesan Log Sederhana (SPL) untuk konsumsi data.
           config.setQuery("* | where cast(body_bytes_sent as bigint) > 14000");
           ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), 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 dari 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, waktu: 1635629778, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   Tags
       __receive_time__    :    1635629815
   --------
   Log: 0, waktu: 1635629788, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   Tags
       __receive_time__    :    1635629877
   --------
   ......