Simple Log Service：Gunakan protokol Kafka untuk mengunggah log

Dependensi

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.1.0</version>
</dependency>

Contoh Kode

package org.example;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProduceExample {

    public static void main(String[] args) {
        // Konfigurasi.
        Properties props = new Properties();
        String project = "etl-shanghai-b";
        String logstore = "testlog";
        // Atur parameter ini ke true jika Anda ingin konten dari produsen diurai sebagai log JSON.
        boolean parseJson = false;
        // Akun Alibaba Cloud memiliki izin penuh pada semua operasi API. Ini menimbulkan ancaman keamanan yang tinggi. Kami sarankan Anda membuat dan menggunakan Pengguna RAM untuk memanggil operasi API atau melakukan O&M rutin. Untuk membuat Pengguna RAM, masuk ke konsol RAM.
        // Bagian ini memberikan contoh cara menyimpan ID AccessKey dan Rahasia AccessKey ke variabel lingkungan. Anda juga dapat menyimpan ID AccessKey dan Rahasia AccessKey ke file konfigurasi sesuai kebutuhan.
        // Kami sarankan Anda tidak menyimpan ID AccessKey dan Rahasia AccessKey di dalam kode untuk mencegah risiko keamanan.
        String accessKeyID = System.getenv("SLS_ACCESS_KEY_ID");
        String accessKeySecret = System.getenv("SLS_ACCESS_KEY_SECRET");
        String endpoint = "cn-shanghai.log.aliyuncs.com"; // Konfigurasikan parameter ini berdasarkan titik akhir proyek.
        String port = "10012"; // Gunakan port 10012 untuk Internet dan port 10011 untuk jaringan internal.

        String hosts = project + "." + endpoint + ":" + port;
        String topic = logstore;
        if(parseJson) {
            topic = topic + ".json";
        }

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put("security.protocol", "sasl_ssl");
        props.put("sasl.mechanism", "PLAIN");
        props.put("sasl.jaas.config",
                "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" +
                        project + "\" password=\"" + accessKeyID + "#" + accessKeySecret + "\";");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");

        // Buat instance produsen.
        KafkaProducer<String,String> producer = new KafkaProducer<>(props);

        // Kirim rekaman.
        for(int i=0;i<1;i++){
            String content = "{\"msg\": \"Hello World\"}";
            // Jika diperlukan, gunakan metode berikut untuk mengatur timestamp pesan.
            // long timestamp = System.currentTimeMillis();
            // ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, timestamp, null, content);

            ProducerRecord<String, String> record = new ProducerRecord<>(topic, content);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("ERROR: Gagal mengirim pesan: " + exception.getMessage());
                    exception.printStackTrace();
                } else {
                    System.out.println("Pesan berhasil dikirim ke topik: " + metadata.topic() +
                                     ", partisi: " + metadata.partition() +
                                     ", offset: " + metadata.offset() +
                                     ", timestamp: " + metadata.timestamp());
                }
            });
        }

        producer.close();
    }
}

Dependensi

pip install confluent-kafka

Contoh kode

#!/bin/env python3
import time
import os
from confluent_kafka import Producer

def delivery_report(err, msg):
    """ Dipanggil sekali untuk setiap pesan yang diproduksi untuk menunjukkan hasil pengiriman.
        Dipicu oleh poll() atau flush(). """
    if err is not None:
        print('Pengiriman pesan gagal: {}'.format(err))
    else:
        print('Pesan berhasil dikirim ke {} [{}] pada offset {}'.format(msg.topic(), msg.partition(), msg.offset()))

def main():
    project = "etl-shanghai-b"
    logstore = "testlog"
    parse_json = False

    # Dapatkan kredensial dari variabel lingkungan
    access_key_id = os.getenv("SLS_ACCESS_KEY_ID")
    access_key_secret = os.getenv("SLS_ACCESS_KEY_SECRET")
    endpoint = "cn-shanghai.log.aliyuncs.com"
    port = "10012" # Gunakan port 10012 untuk Internet dan port 10011 untuk jaringan internal.

    hosts = f"{project}.{endpoint}:{port}"
    topic = logstore
    if parse_json:
        topic = topic + ".json"

    # Konfigurasikan produsen Kafka
    conf = {
        'bootstrap.servers': hosts,
        'security.protocol': 'sasl_ssl',
        'sasl.mechanisms': 'PLAIN',
        'sasl.username': project,
        'sasl.password': f"{access_key_id}#{access_key_secret}",
        'enable.idempotence': False,
    }

    # Buat instance produsen
    producer = Producer(conf)

    # Kirim pesan
    content = "{\"msg\": \"Hello World\"}"
    producer.produce(topic=topic,
                     value=content.encode('utf-8'),
                     #timestamp=int(time.time() * 1000),  # (Opsional) Setel timestamp rekaman dalam milidetik.
                     callback=delivery_report)

    # Tunggu pesan yang belum selesai dikirim dan laporan pengiriman dipicu.
    producer.flush()

if __name__ == '__main__':
    main()

Dependensi

go get github.com/confluentinc/confluent-kafka-go/kafka

Contoh kode

package main

import (
	"fmt"
	"log"
	"os"
//	"time"

	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	project := "etl-shanghai-b"
	logstore := "testlog"
	parseJson := false

	// Dapatkan kredensial dari variabel lingkungan
	accessKeyID := os.Getenv("SLS_ACCESS_KEY_ID")
	accessKeySecret := os.Getenv("SLS_ACCESS_KEY_SECRET")
	endpoint := "cn-shanghai.log.aliyuncs.com"
	port := "10012" // Gunakan port 10012 untuk Internet dan port 10011 untuk jaringan internal.

	hosts := fmt.Sprintf("%s.%s:%s", project, endpoint, port)
	topic := logstore
	if parseJson {
		topic = topic + ".json"
	}

	// Konfigurasikan produsen Kafka
	config := &kafka.ConfigMap{
		"bootstrap.servers":  hosts,
		"security.protocol":  "sasl_ssl",
		"sasl.mechanisms":    "PLAIN",
		"sasl.username":      project,
		"sasl.password":      accessKeyID + "#" + accessKeySecret,
		"enable.idempotence": false,
	}

	// Buat instance produsen
	producer, err := kafka.NewProducer(config)
	if err != nil {
		log.Fatalf("Gagal membuat produsen: %v", err)
	}
	defer producer.Close()

	// Kirim pesan dalam batch.
	messages := []string{
		"{\"msg\": \"Hello World 1\"}",
		"{\"msg\": \"Hello World 2\"}",
		"{\"msg\": \"Hello World 3\"}",
	}

	for _, content := range messages {
		err := producer.Produce(&kafka.Message{
			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
			Value:          []byte(content),
			//Timestamp:      time.Now(), // Setel waktu jika diperlukan.
		}, nil)

		if err != nil {
			log.Printf("Gagal memproduksi pesan: %v", err)
		}
	}

	// Aktifkan goroutine untuk mendengarkan apakah produsen berhasil mengirim pesan.
	go func() {
		for e := range producer.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					fmt.Printf("Pengiriman gagal: %v\n", ev.TopicPartition.Error)
				} else {
					fmt.Printf("Pesan terkirim ke topik %s [%d] pada offset %v\n",
						*ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
				}
			}
		}
	}()

	producer.Flush(5 * 1000)
}