Simple Log Service:Mengonsumsi log dengan kelompok konsumen

SDK

Kode berikut membuat kelompok konsumen:

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 memperoleh 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 titik akhir untuk Simple Log Service. Contoh ini menggunakan titik akhir Wilayah Tiongkok (Hangzhou). Ganti dengan titik akhir yang sebenarnya.
        String host = "https://cn-hangzhou.log.aliyuncs.com";

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

        try {
            // Tetapkan nama kelompok konsumen.
            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 kelompok konsumen, lihat Use the Java SDK to manage consumer groups dan Use Simple Log Service SDK for Python to manage consumer groups.

API

Untuk membuat kelompok konsumen menggunakan API, lihat CreateConsumerGroup.

Untuk memverifikasi bahwa kelompok konsumen telah dibuat, lihat ListConsumerGroup.

CLI

Untuk membuat kelompok konsumen menggunakan CLI, lihat create_consumer_group.

Untuk memverifikasi bahwa kelompok konsumen 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 prosesor.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
       // Logika utama untuk mengonsumsi data. Semua pengecualian harus ditangani dalam metode ini. Jangan melempar pengecualian 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 pekerja berhenti secara tak terduga, pekerja baru akan melanjutkan dari checkpoint terakhir, 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 pekerja 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 prosesor 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 konsumen. 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 {
       // Titik akhir Simple Log Service. Ganti dengan titik akhir 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 kelompok konsumen 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 konsumen, yang harus unik dalam satu kelompok konsumen. Untuk menyeimbangkan konsumsi di beberapa mesin, Anda dapat menggunakan pengenal unik seperti alamat IP mesin sebagai nama konsumen.
           // 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() pekerja untuk menghentikan instance konsumen dan thread terkaitnya.
           worker.shutdown();
           // ClientWorker membuat tugas asinkron. Setelah shutdown(), tunggu tugas-tugas tersebut selesai secara elegan. Disarankan untuk memberi jeda selama 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
   --------
   ......