Wujudkan teknologi caching masa nyata berdasarkan baris gilir mesej Kafka di Golang.

Dengan pembangunan berterusan teknologi Internet dan pengembangan berterusan senario aplikasi, teknologi caching masa nyata semakin menjadi kemahiran penting untuk syarikat Internet. Sebagai kaedah teknologi caching masa nyata, baris gilir mesej semakin digemari oleh pembangun dalam aplikasi praktikal. Artikel ini terutamanya memperkenalkan cara mewujudkan teknologi caching masa nyata berdasarkan baris gilir mesej Kafka di Golang.

Apakah baris gilir mesej Kafka?

Kafka ialah sistem pemesejan teragih yang dibangunkan oleh LinkedIn dan boleh mengendalikan berpuluh juta mesej. Ia mempunyai ciri daya pemprosesan tinggi, kependaman rendah, ketahanan dan kebolehpercayaan yang tinggi. Kafka mempunyai tiga komponen utama: pengeluar, pengguna dan topik Antaranya, pengeluar dan pengguna adalah bahagian teras Kafka.

Pengeluar menghantar mesej ke topik yang ditentukan, dan juga boleh menentukan partition dan kunci. Pengguna menerima mesej yang sepadan daripada topik tersebut. Di Kafka, pengeluar dan pengguna adalah bebas dan tidak mempunyai pergantungan antara satu sama lain Mereka hanya berinteraksi antara satu sama lain dengan berkongsi topik yang sama. Seni bina ini melaksanakan pemesejan teragih dan menyelesaikan keperluan baris gilir mesej secara berkesan dalam pelbagai senario perniagaan.

Gabungan Golang dan Kafka

Golang ialah bahasa pengaturcaraan yang popular dan cekap sejak beberapa tahun kebelakangan ini, ia semakin digunakan secara meluas kerana keselarasannya yang tinggi dan prestasi tinggi. Ia mempunyai kelebihan yang wujud apabila digabungkan dengan baris gilir mesej, kerana di Golang, bilangan goroutine mempunyai perhubungan satu dengan satu dengan bilangan utas kernel, yang bermaksud bahawa Golang boleh mengendalikan tugas serentak berskala besar dengan cekap dan lancar, manakala Kafka boleh Mengedarkan pelbagai mesej kepada nod broker yang berbeza mengikut peraturan partition yang boleh disesuaikan untuk mencapai pengembangan mendatar.

Dengan menggunakan sarama perpustakaan Kafka pihak ketiga di Golang, kami boleh melaksanakan interaksi dengan Kafka dengan mudah. Langkah-langkah pelaksanaan khusus adalah seperti berikut:

1 Memperkenalkan perpustakaan sarama ke dalam projek Golang:

import "github.com/Shopify/sarama"

2 Buat contoh pengirim mesej (Pengeluar):

config := sarama.NewConfig()
config.Producer.Return.Successes = true
producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)

Antaranya, NewConfig() digunakan untuk membuat contoh fail konfigurasi baharu.Kejayaan bermakna maklumat kejayaan akan dikembalikan apabila setiap mesej berjaya dihantar untuk mencipta contoh rentetan mewakili Broker dalam kelompok Kafka Alamat IP dan nombor port nod.

3. Hantar mesej:

msg := &sarama.ProducerMessage{
  Topic: "test-topic",
  Value: sarama.StringEncoder("hello world"),
}
producer.Input() <- msg

Antaranya, ProducerMessage mewakili struktur mesej, Topik mewakili topik yang menjadi milik mesej, dan Nilai mewakili kandungan mesej.

4. Buat contoh mesej pengguna (Pengguna):

config := sarama.NewConfig()
config.Consumer.Return.Errors = true
consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)

Antaranya, NewConfig() digunakan untuk membuat contoh fail konfigurasi baharu dan Return.Error bermakna setiap kali mesej digunakan, Mengembalikan mesej ralat kegagalan penggunaan NewConsumer() digunakan untuk mencipta contoh pengguna.

5. Gunakan mesej:

partitionConsumer, err := consumer.ConsumePartition("test-topic", 0, sarama.OffsetNewest)
for msg := range partitionConsumer.Messages() {
  fmt.Printf("Consumed message: %s
", string(msg.Value))
  partitionConsumer.MarkOffset(msg, "") // 确认消息已被消费
}

Antaranya, ConsumePartition() digunakan untuk menentukan topik, partition dan lokasi penggunaan (mesej terkini atau mesej tertua), dan Messages() digunakan untuk dapatkan daripada Mesej yang digunakan dalam topik. Selepas menggunakan mesej, kita perlu menggunakan kaedah MarkOffset() untuk mengesahkan bahawa mesej telah digunakan.

Pelaksanaan cache masa nyata Kafka

Di Golang, adalah sangat mudah untuk mewujudkan cache masa nyata melalui baris gilir mesej Kafka. Kami boleh mencipta modul pengurusan cache dalam projek, menukar kandungan cache ke dalam struktur mesej yang sepadan mengikut keperluan sebenar, menghantar mesej ke topik yang ditentukan dalam kelompok Kafka melalui pengeluar, dan menunggu pengguna untuk menggunakan mesej daripada topik dan teruskan.

Berikut ialah langkah pelaksanaan khusus:

1 Tentukan struktur cache dan pembolehubah cache dalam projek:

type Cache struct {
  Key   string
  Value interface{}
}

var cache []Cache

Antaranya, Kunci mewakili kunci cache ( Key), Nilai mewakili nilai cache (Nilai).

2. Tukar cache ke dalam struktur mesej yang sepadan:

type Message struct {
  Operation string // 操作类型（Add/Delete/Update）
  Cache     Cache  // 缓存内容
}

func generateMessage(operation string, cache Cache) Message {
  return Message{
    Operation: operation,
    Cache:     cache,
  }
}

Antaranya, Mesej mewakili struktur mesej, Operasi mewakili jenis operasi cache, dan generateMessage() digunakan untuk mengembalikan Mesej contoh.

3. Tulis penerbit dan hantar kandungan cache sebagai mesej ke topik yang ditentukan:

func producer(messages chan *sarama.ProducerMessage) {
  config := sarama.NewConfig()
  config.Producer.Return.Successes = true
  producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)
  if err != nil {
    panic(err)
  }

  for {
    select {
    case msg := <-messages:
      producer.Input() <- msg
    }
  }
}

func pushMessage(operation string, cache Cache, messages chan *sarama.ProducerMessage) {
  msg := sarama.ProducerMessage{
    Topic: "cache-topic",
    Value: sarama.StringEncoder(generateMessage(operation, cache)),
  }
  messages <- &msg
}

Antaranya, producer() digunakan untuk mencipta contoh pengeluar dan menunggu mesej masuk daripada saluran paip yang akan dihantar , pushMessage() digunakan untuk menukar kandungan cache kepada contoh Mesej dan menghantarnya ke topik yang ditentukan menggunakan pengeluar.

4 Tulis pengguna, dengar topik yang ditentukan dan lakukan operasi yang sepadan apabila mesej tiba:

func consumer() {
  config := sarama.NewConfig()
  config.Consumer.Return.Errors = true
  consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
  if err != nil {
    panic(err)
  }

  partitionConsumer, err := consumer.ConsumePartition("cache-topic", 0, sarama.OffsetNewest)
  if err != nil {
    panic(err)
  }

  for msg := range partitionConsumer.Messages() {
    var message Message
    err := json.Unmarshal(msg.Value, &message)
    if err != nil {
      fmt.Println("Failed to unmarshal message: ", err.Error())
      continue
    }

    switch message.Operation {
    case "Add":
      cache = append(cache, message.Cache)
    case "Delete":
      for i, c := range cache {
        if c.Key == message.Cache.Key {
          cache = append(cache[:i], cache[i+1:]...)
          break
        }
      }
    case "Update":
      for i, c := range cache {
        if c.Key == message.Cache.Key {
          cache[i] = message.Cache
          break
        }
      }
    }
    partitionConsumer.MarkOffset(msg, "") // 确认消息已被消费
  }
}

Antaranya, pengguna() digunakan untuk mencipta contoh pengguna dan mendengar topik yang ditentukan, gunakan Fungsi json.Unmarshal() menghuraikan medan Nilai mesej ke dalam struktur Mesej, dan kemudian melaksanakan operasi caching yang sepadan berdasarkan medan Operasi. Selepas menggunakan mesej, kita perlu menggunakan kaedah MarkOffset() untuk mengesahkan bahawa mesej telah digunakan.

Melalui langkah di atas, kami telah berjaya mewujudkan teknologi caching masa nyata berdasarkan baris gilir mesej Kafka menggunakan sarama, perpustakaan Kafka di Golang. Dalam aplikasi praktikal, kita boleh memilih konfigurasi gugusan Kafka yang berbeza dan peraturan partition mengikut keperluan sebenar untuk menghadapi pelbagai senario perniagaan secara fleksibel.

Atas ialah kandungan terperinci Wujudkan teknologi caching masa nyata berdasarkan baris gilir mesej Kafka di Golang.. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!