Richten Sie in Golang eine Echtzeit-Caching-Technologie basierend auf der Kafka-Nachrichtenwarteschlange ein.

Mit der kontinuierlichen Weiterentwicklung der Internet-Technologie und der kontinuierlichen Erweiterung der Anwendungsszenarien ist die Echtzeit-Caching-Technologie zunehmend zu einer wesentlichen Fähigkeit für Internetunternehmen geworden. Als Methode der Echtzeit-Caching-Technologie wird die Nachrichtenwarteschlange von Entwicklern in praktischen Anwendungen zunehmend bevorzugt. In diesem Artikel wird hauptsächlich die Einrichtung einer Echtzeit-Caching-Technologie basierend auf der Kafka-Nachrichtenwarteschlange in Golang vorgestellt.

Was ist die Kafka-Nachrichtenwarteschlange?

Kafka ist ein von LinkedIn entwickeltes verteiltes Nachrichtensystem, das zig Millionen Nachrichten verarbeiten kann. Es zeichnet sich durch hohen Durchsatz, geringe Latenz, Haltbarkeit und hohe Zuverlässigkeit aus. Kafka besteht aus drei Hauptkomponenten: Produzenten, Konsumenten und Themen. Unter ihnen sind Produzenten und Konsumenten die Kernbestandteile von Kafka.

Der Produzent sendet Nachrichten an das angegebene Thema und kann auch die Partition und den Schlüssel angeben. Verbraucher erhalten entsprechende Nachrichten zum Thema. In Kafka sind Produzenten und Konsumenten unabhängig und haben keine Abhängigkeiten voneinander. Sie interagieren nur miteinander, indem sie dasselbe Thema teilen. Diese Architektur implementiert verteiltes Messaging und löst effektiv die Anforderungen an Nachrichtenwarteschlangen in verschiedenen Geschäftsszenarien.

Die Kombination aus Golang und Kafka

Golang ist in den letzten Jahren eine beliebte und effiziente Programmiersprache. Aufgrund seiner hohen Parallelität, hohen Leistung und anderen Eigenschaften wird es immer häufiger verwendet. Es hat den inhärenten Vorteil der Kombination mit Nachrichtenwarteschlangen, da in Golang die Anzahl der Goroutinen in einer Eins-zu-eins-Beziehung zur Anzahl der Kernel-Threads steht, was bedeutet, dass Golang große gleichzeitige Aufgaben effizient und reibungslos bewältigen kann Kafka kann verschiedene Nachrichten gemäß anpassbaren Partitionsregeln an verschiedene Brokerknoten verteilen, um eine horizontale Erweiterung zu erreichen.

Durch die Verwendung der Kafka-Bibliothek sarama eines Drittanbieters in Golang können wir die Interaktion mit Kafka problemlos implementieren. Die spezifischen Implementierungsschritte sind wie folgt:

1. Führen Sie die Sarama-Bibliothek in das Golang-Projekt ein:

import "github.com/Shopify/sarama"

2. Erstellen Sie eine Instanz des Nachrichtensenders (Produzent):

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

Unter anderem wird NewConfig() verwendet, um eine neue zu erstellen Konfigurationsdateiinstanz, Return.Successes bedeutet, dass Erfolgsinformationen zurückgegeben werden, wenn jede Nachricht erfolgreich gesendet wird NewAsyncProducer() Das String-Array in den Parametern stellt die IP-Adresse und die Portnummer des Broker-Knotens dar der Kafka-Cluster.

3. Senden Sie eine Nachricht:

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

Dabei repräsentiert ProducerMessage die Nachrichtenstruktur, Topic das Thema, zu dem die Nachricht gehört, und Value den Nachrichteninhalt.

4. Erstellen Sie eine Nachrichtenkonsumenteninstanz (Consumer):

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

Unter anderem wird NewConfig() verwendet, um eine neue Konfigurationsdateiinstanz zu erstellen. Return.Errors bedeutet, dass bei jeder Nachricht eine Fehlermeldung über den Verbrauchsfehler zurückgegeben wird verbraucht. NewConsumer() wird verwendet, um eine Verbraucherinstanz zu erstellen.

5. Nachrichten konsumieren:

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, "") // 确认消息已被消费
}

Unter diesen wird ConsumePartition() verwendet, um das Thema, die Partition und den Verbrauchsort (letzte Nachricht oder älteste Nachricht) des Verbrauchs anzugeben, und Messages() wird verwendet, um die verbrauchten Nachrichten abzurufen Thema. Nachdem wir eine Nachricht konsumiert haben, müssen wir die Methode MarkOffset() verwenden, um zu bestätigen, dass die Nachricht konsumiert wurde.

Kafka-Echtzeit-Cache-Implementierung

In Golang ist es sehr praktisch, einen Echtzeit-Cache über die Kafka-Nachrichtenwarteschlange einzurichten. Wir können im Projekt ein Cache-Verwaltungsmodul erstellen, den Cache-Inhalt entsprechend den tatsächlichen Anforderungen in die entsprechende Nachrichtenstruktur umwandeln, die Nachricht über den Produzenten an das angegebene Thema im Kafka-Cluster senden und darauf warten, dass der Verbraucher die Nachricht konsumiert das Thema und befassen Sie sich damit.

Das Folgende sind die spezifischen Implementierungsschritte:

1 Definieren Sie eine Cache-Struktur und eine Cache-Variable im Projekt:

type Cache struct {
  Key   string
  Value interface{}
}

var cache []Cache

Unter diesen stellt Key den zwischengespeicherten Schlüssel (Key) und Value den zwischengespeicherten Wert (Value) dar. .

2. Konvertieren Sie den Cache in die entsprechende Nachrichtenstruktur:

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

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

Unter diesen stellt Message die Nachrichtenstruktur dar, Operation stellt den Cache-Operationstyp dar und genericMessage() wird verwendet, um eine Nachrichteninstanz zurückzugeben.

3. Schreiben Sie einen Produzenten und senden Sie den zwischengespeicherten Inhalt als Nachricht an das angegebene Thema:

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
}

Unter anderem wird Producer() verwendet, um eine Producer-Instanz zu erstellen und auf das Senden der von der Pipeline eingehenden Nachricht zu warten pushMessage() wird zum Zwischenspeichern verwendet. Der Inhalt wird in eine Nachrichteninstanz konvertiert und mithilfe des Produzenten an das angegebene Thema gesendet.

4. Schreiben Sie einen Verbraucher, hören Sie sich das angegebene Thema an und führen Sie entsprechende Vorgänge aus, wenn die Nachricht eintrifft:

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, "") // 确认消息已被消费
  }
}

Unter anderem wird Consumer () verwendet, um eine Verbraucherinstanz zu erstellen, das angegebene Thema abzuhören und den JSON zu verwenden. Unmarshal()-Funktion zum Entmarshalieren der Nachricht. Das Feld „Wert“ wird in eine Nachrichtenstruktur geparst, und dann wird die entsprechende Cache-Operation basierend auf dem Feld „Operation“ ausgeführt. Nachdem wir eine Nachricht konsumiert haben, müssen wir die Methode MarkOffset() verwenden, um zu bestätigen, dass die Nachricht konsumiert wurde.

Durch die oben genannten Schritte haben wir mithilfe von sarama, der Kafka-Bibliothek in Golang, erfolgreich eine Echtzeit-Caching-Technologie basierend auf der Kafka-Nachrichtenwarteschlange etabliert. In praktischen Anwendungen können wir entsprechend den tatsächlichen Anforderungen unterschiedliche Kafka-Clusterkonfigurationen und Partitionsregeln auswählen, um verschiedene Geschäftsszenarien flexibel zu bewältigen.

Das obige ist der detaillierte Inhalt vonRichten Sie in Golang eine Echtzeit-Caching-Technologie basierend auf der Kafka-Nachrichtenwarteschlange ein.. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!