Simple Log Service:Mengonsumsi log dengan kelompok konsumen

Overview

Sebuah Logstore berisi beberapa shard. Saat Anda menggunakan consumer group untuk mengonsumsi data, shard diberikan kepada konsumen dalam grup tersebut sebagai berikut:

• Dalam satu consumer group, setiap shard hanya diberikan kepada satu konsumen.

• Dalam satu consumer group, satu konsumen dapat diberikan beberapa shard.

Saat konsumen baru bergabung ke dalam consumer group, shard akan ditugaskan ulang untuk menyeimbangkan beban. Penugasan ulang ini mengikuti prinsip yang dijelaskan di atas.

Key concepts

Prerequisites

• Proyek dan logstore Standard telah dibuat, serta log telah dikumpulkan. Untuk informasi lebih lanjut, lihat Manage projects, Create a basic logstore, dan Data collection.

• Configure access credentials.

• Initialize Simple Log Service SDK for Java.

Step 1: Create a consumer group

Bagian ini menunjukkan cara membuat consumer group menggunakan SDK, API, atau CLI.

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

        // Buat klien 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;
        }
    }
}

How it works

Saat konsumen pertama kali dijalankan, SDK akan membuat consumer group jika belum ada. Pengaturan start consumption checkpoint menentukan posisi awal dan hanya berlaku saat consumer group pertama kali dibuat. Pada restart berikutnya, konsumen melanjutkan dari checkpoint konsumsi terakhir yang disimpan di server. Contohnya:

• LogHubConfig.ConsumePosition.BEGIN_CURSOR: Consumer group memulai konsumsi dari log paling awal yang tersedia di Logstore.

• LogHubConfig.ConsumePosition.END_CURSOR: Consumer group hanya memproses log baru yang tiba setelah konsumen dijalankan.

Consumption examples

Step 3: View consumer group status

Bagian ini menjelaskan dua cara untuk melihat status consumer group.

Related operations

• Authorize a RAM user

  Untuk mengelola consumer group dengan RAM user, berikan izin yang diperlukan kepada pengguna tersebut. Untuk informasi lebih lanjut, lihat Create a RAM user and grant permissions.

  Tabel berikut mencantumkan tindakan yang diperlukan.

  Action	Description	Resource
  log:GetCursorOrData (GetCursor)	Mendapatkan cursor berdasarkan waktu tertentu.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}
  log:CreateConsumerGroup (CreateConsumerGroup)	Membuat consumer group untuk Logstore tertentu.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}/consumergroup/${consumerGroupName}
  log:ListConsumerGroup (ListConsumerGroup)	Menampilkan semua consumer group untuk Logstore tertentu.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}/consumergroup/*
  log:ConsumerGroupUpdateCheckPoint (ConsumerGroupUpdateCheckPoint)	Memperbarui checkpoint shard untuk consumer group tertentu.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}/consumergroup/${consumerGroupName}
  log:ConsumerGroupHeartBeat (ConsumerGroupHeartBeat)	Mengirim heartbeat dari konsumen ke server.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}/consumergroup/${consumerGroupName}
  log:UpdateConsumerGroup (UpdateConsumerGroup)	Memodifikasi properti consumer group tertentu.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}/consumergroup/${consumerGroupName}
  log:GetConsumerGroupCheckPoint (GetCheckPoint)	Mendapatkan checkpoint satu atau semua shard untuk consumer group tertentu.	acs:log:${regionName}:${projectOwnerAliUid}:project/${projectName}/logstore/${LogStoreName}/consumergroup/${consumerGroupName}

  Sebagai contoh, untuk memberikan izin kepada RAM user untuk consumer group dengan detail berikut, gunakan kebijakan di bawah ini.

  • Akun Alibaba Cloud: 174649****602745.

  • ID wilayah: cn-hangzhou.

  • Nama proyek: project-test.

  • Nama Logstore: logstore-test.

  • Nama consumer group: consumergroup-test.

  Contoh kebijakan:

  {
    "Version": "1",
    "Statement": [
      {
        "Effect": "Allow",
        "Action": [
          "log:GetCursorOrData"
        ],
        "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test"
      },
      {
        "Effect": "Allow",
        "Action": [
          "log:CreateConsumerGroup",
          "log:ListConsumerGroup"
        ],
        "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test/consumergroup/*"
      },
      {
        "Effect": "Allow",
        "Action": [
          "log:ConsumerGroupUpdateCheckPoint",
          "log:ConsumerGroupHeartBeat",
          "log:UpdateConsumerGroup",
          "log:GetConsumerGroupCheckPoint"
        ],
        "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test/consumergroup/consumergroup-test"
      }
    ]
  }

• Troubleshooting

  Konfigurasikan Log4j untuk aplikasi konsumen Anda agar mencatat pengecualian dari consumer group. Hal ini membantu Anda menemukan dan mendiagnosis masalah. Kode berikut menunjukkan contoh konfigurasi log4j.properties:

  log4j.rootLogger = info,stdout
  log4j.appender.stdout = org.apache.log4j.ConsoleAppender
  log4j.appender.stdout.Target = System.out
  log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
  log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

  Setelah mengonfigurasi Log4j, Anda akan melihat pesan pengecualian seperti contoh berikut saat menjalankan aplikasi konsumen:

  [WARN ] 2018-03-14 12:01:52,747 method:com.aliyun.openservices.loghub.client.LogHubConsumer.sampleLogError(LogHubConsumer.java:159)
  com.aliyun.openservices.log.exception.LogException: Invalid loggroup count, (0,1000]

• Consume data from a specific time

  // Parameter consumerStartTimeInSeconds menentukan waktu mulai konsumsi.
  public LogHubConfig(String consumerGroupName, 
                        String consumerName, 
                        String loghubEndPoint,
                        String project, String logStore,
                        String accessId, String accessKey,
                        int consumerStartTimeInSeconds);
  
  // Parameter position adalah enumerasi. LogHubConfig.ConsumePosition.BEGIN_CURSOR memulai konsumsi dari data paling awal, dan LogHubConfig.ConsumePosition.END_CURSOR memulai dari data terbaru.
  public LogHubConfig(String consumerGroupName, 
                        String consumerName, 
                        String loghubEndPoint,
                        String project, String logStore,
                        String accessId, String accessKey,
                        ConsumePosition position);

  Catatan

  • Gunakan konstruktor yang sesuai dengan kebutuhan konsumsi Anda.

  • Jika checkpoint sudah disimpan di server, konsumsi dilanjutkan dari checkpoint yang tersimpan, terlepas dari pengaturan lainnya.

  • Secara default, Simple Log Service memprioritaskan checkpoint untuk konsumsi data. Jika Anda menentukan waktu mulai, pastikan nilai consumerStartTimeInSeconds berada dalam periode waktu hidup (TTL). Jika tidak, pengaturan tersebut tidak berpengaruh.

• Reset a checkpoint

  public static void updateCheckpoint() throws Exception {
          Client client = new Client(host, accessId, accessKey);
          // Stempel waktu harus berupa stempel waktu UNIX dalam detik. Jika stempel waktu Anda dalam milidetik, bagi dengan 1000 seperti pada contoh.
          long timestamp = Timestamp.valueOf("2017-11-15 00:00:00").getTime() / 1000;
          ListShardResponse response = client.ListShard(new ListShardRequest(project, logStore));
          for (Shard shard : response.GetShards()) {
              int shardId = shard.GetShardId();
              String cursor = client.GetCursor(project, logStore, shardId, timestamp).GetCursor();
              client.UpdateCheckPoint(project, logStore, consumerGroup, shardId, cursor);
          }
      }

References

• API

  Actions	API
  Create consumer group	CreateConsumerGroup
  List consumer groups	ListConsumerGroup
  Delete consumer group	DeleteConsumerGroup
  Update consumer group	UpdateConsumerGroup
  Send consumer heartbeat	ConsumerGroupHeartBeat
  Get consumer group checkpoint	GetCheckPoint
  Update consumer group checkpoint	ConsumerGroupUpdateCheckPoint

• SDK

  Language	Documentation
  Java	Manage consumer groups with the Java SDK Consume data using consumer groups
  Python	Manage consumer groups with the Python SDK Consume data using consumer groups

• CLI

  Actions	CLI
  Create consumer group	create_consumer_group
  List consumer groups	list_consumer_group
  Update consumer group	update_consumer_group
  Delete consumer group	delete_consumer_group
  Get consumer group checkpoint	get_check_point
  Update consumer group checkpoint	update_check_point