Amalan go-zero dan Kafka+Avro: membina sistem pemprosesan data interaktif berprestasi tinggi

Dalam beberapa tahun kebelakangan ini, dengan peningkatan data besar dan komuniti sumber terbuka yang aktif, semakin banyak perusahaan telah mula mencari sistem pemprosesan data interaktif berprestasi tinggi untuk memenuhi keperluan data yang semakin meningkat. Dalam gelombang peningkatan teknologi ini, go-zero dan Kafka+Avro sedang diberi perhatian dan diterima pakai oleh semakin banyak perusahaan.

go-zero ialah rangka kerja mikroperkhidmatan yang dibangunkan berdasarkan bahasa Golang Ia mempunyai ciri-ciri prestasi tinggi, kemudahan penggunaan, pengembangan mudah dan penyelenggaraan yang mudah Ia direka untuk membantu perusahaan membina aplikasi perkhidmatan mikro yang cekap sistem. Pertumbuhan pesatnya adalah disebabkan oleh prestasi cemerlang dan kecekapan pembangunan tinggi Golang itu sendiri, serta lelaran dan pengoptimuman berterusan pasukan go-zero.

Kafka ialah sistem pemprosesan strim teragih yang dibangunkan oleh Apache Ia mempunyai ciri-ciri ketersediaan tinggi dan daya pemprosesan yang tinggi Ia merupakan salah satu baris gilir mesej yang paling popular dalam ekosistem data besar semasa. Avro ialah alat siri data yang dibangunkan oleh Apache, yang boleh menukar aliran data ke dalam format binari, dengan itu meningkatkan pemampatan data dan kecekapan penghantaran. Ia juga boleh menyokong peningkatan dan penukaran format data.

Dalam artikel ini, kami akan memperkenalkan cara menggabungkan go-zero dan Kafka+Avro untuk membina sistem pemprosesan data interaktif berprestasi tinggi. Proses praktikal khusus adalah seperti berikut:

1. Sepadukan pelanggan Kafka

Pertama, kita perlu menyepadukan pelanggan Kafka dalam perkhidmatan go-zero. go-zero menyediakan pakej Kafka yang boleh berinteraksi dengan Kafka dengan mudah.

Kami hanya perlu memperkenalkan pakej Kafka ke dalam projek dan mengkonfigurasi parameter Kafka dalam fail konfigurasi untuk mencapai sambungan dan interaksi data dengan Kafka. Berikut ialah contoh konfigurasi Kafka:

[kafka]
addrs = ["localhost:9092"]
version = "2.0.0"
maxMessageBytes = 10000000

Dalam logik perniagaan tertentu, kami boleh menggunakan API pengeluar dan pengguna yang disediakan oleh Kafka untuk menghantar dan menerima data. Berikut ialah contoh pengeluar Kafka:

var (
    topic = "test"
)

func (s *Service) Produce(msg []byte) error {
    p, err := kafka.NewProducer(s.cfg.Kafka)
    if err != nil {
        return err
    }
    defer p.Close()

    return p.Send(context.TODO(), &kafka.Message{
        Key:   []byte(topic),
        Value: msg,
    })
}

Dalam contoh di atas, kami telah mencipta topik Kafka bernama "ujian" dan apabila kaedah Hasilkan dipanggil, data dihantar ke topik.

2. Pesirilan Avro Bersepadu

Seterusnya, kita perlu menukar data ke dalam format Avro untuk penyirian dan penyahserilan. go-zero menyediakan pakej Avro dan menyokong penjanaan kod. Dengan mentakrifkan fail Skema, kami boleh menjana kod Go yang sepadan untuk mengekod dan menyahkod data Avro.

Berikut ialah contoh konfigurasi Avro Schema:

{
    "namespace": "com.example",
    "type": "record",
    "name": "User",
    "fields": [
        {
            "name": "name",
            "type": "string"
        },
        {
            "name": "age",
            "type": "int"
        }
    ]
}

Dengan melaksanakan arahan berikut, fail Go yang sepadan boleh dijana secara automatik:

$ go run github.com/gogo/protobuf/protoc-gen-gogofaster --proto_path=./ example.proto --gogofaster_out

Dalam fail Go yang dijana, kita boleh melihat kepada hubungan pemetaan antara jenis medan Avro dan jenis data Go yang sepadan, dengan itu merealisasikan pensirilan dan penyahsirilan data.

3. Membina sistem pemprosesan data interaktif

Selepas menyepadukan Kafka dan Avro, kami boleh mula membina sistem pemprosesan data interaktif berprestasi tinggi. Kami boleh menggunakan Kafka sebagai pusat storan data dan mewujudkan berbilang partition di dalamnya untuk mencapai storan dan pemprosesan data yang diedarkan.

Untuk setiap partition, kami boleh mencipta kumpulan pengguna untuk mencapai pemprosesan selari dan pengimbangan beban data. Pada masa yang sama, kami boleh menggunakan kumpulan coroutine dan saluran penyegerakan yang disediakan oleh go-zero untuk mengoptimumkan prestasi serentak pemprosesan data.

Berikut ialah contoh sistem pemprosesan data interaktif:

// 创建消费组
group, err := kafka.NewGroup(s.cfg.Kafka, "test", kafka.WithGroupID("test-group"))
if err != nil {
    return nil, err
}
// 创建消费者
consumer, err := group.NewConsumer(context.Background(), []string{"test"})
if err != nil {
    return nil, err
}
// 启动并发协程
for i := 0; i < s.cfg.WorkerNum; i++ {
    go func() {
        for {
            select {
                // 从同步通道中获取新消息
                case msg := <-msgs:
                    if err := s.processMsg(msg); err != nil {
                        log.Errorf("failed to process message(%v): %v", msg.Value, err)
                    }
                }
        }
    }()
}
// 消费数据
for {
    m, err := consumer.FetchMessage(context.Background())
    if err != nil {
        log.Errorf("failed to fetch message: %v", err)
        continue
    }
    // 将新消息发送到同步通道中
    msgs <- m
}

Dalam contoh di atas, kami mencipta kumpulan pengguna "kumpulan ujian" dan mencipta pengguna yang sepadan. Semasa pemprosesan, kami mula-mula memulakan berbilang coroutine serentak untuk mencapai pemprosesan data selari. Apabila mesej baharu diterima, kami menghantarnya ke saluran segerak dan menggunakan kumpulan coroutine untuk pemprosesan tak segerak.

Melalui pembinaan di atas, kami berjaya menyepadukan go-zero, Kafka dan Avro untuk melaksanakan sistem pemprosesan data interaktif berprestasi tinggi. Menggunakan sistem jenis ini boleh mengendalikan data besar-besaran dengan mudah dan meningkatkan kecekapan pemprosesan dan analisis data.

Atas ialah kandungan terperinci Amalan go-zero dan Kafka+Avro: membina sistem pemprosesan data interaktif berprestasi tinggi. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!