Golang で Kafka メッセージキューに基づくリアルタイムキャッシュテクノロジを確立します。-Golang-php.cn

インターネットテクノロジの継続的な開発とアプリケーションシナリオの継続的な拡大により、リアルタイムキャッシュテクノロジはインターネット企業にとってますます不可欠なスキルになってきています。リアルタイムキャッシュテクノロジの方法として、メッセージキューは実際のアプリケーションで開発者の間でますます好まれています。この記事では主に、Golang で Kafka メッセージキューに基づくリアルタイムキャッシュ技術を確立する方法を紹介します。

Kafka メッセージキューとは何ですか?

Kafka は、LinkedIn によって開発された分散メッセージングシステムで、数千万のメッセージを処理できます。高スループット、低遅延、耐久性、高信頼性という特徴があります。 Kafka にはプロデューサー、コンシューマー、トピックという 3 つの主要なコンポーネントがあり、このうちプロデューサーとコンシューマーが Kafka の中核部分です。

プロデューサは、指定されたトピックにメッセージを送信し、パーティションとキー (Key) を指定することもできます。コンシューマはトピックから対応するメッセージを受け取ります。 Kafka では、プロデューサーとコンシューマーは独立しており、互いに依存関係はなく、同じトピックを共有することによってのみ相互作用します。このアーキテクチャは分散メッセージ配信を実装し、さまざまなビジネスシナリオにおけるメッセージキューの要件を効果的に解決します。

Golang と Kafka の組み合わせ

Golang は、近年人気が高まっている効率的なプログラミング言語であり、高い同時実行性、高いパフォーマンスなどの特徴を備え、ますます広く使用されています。 Golang ではゴルーチンの数がカーネルスレッドの数と 1 対 1 の関係にあるため、メッセージキューと組み合わせるという固有の利点があります。これは、Golang が大規模な同時タスクを効率的かつスムーズに処理できることを意味します。 Kafka は、水平方向の拡張を実現するために、カスタマイズ可能なパーティションルールに従ってさまざまなメッセージを異なるブローカーノードに配信できます。

Golang でサードパーティの Kafka ライブラリ sarama を使用すると、Kafka との対話を簡単に実装できます。具体的な実装手順は次のとおりです:

1. Golang プロジェクトに sarama ライブラリを導入します:

import "github.com/Shopify/sarama"

ログイン後にコピー

2. メッセージ送信者 (プロデューサー) インスタンスを作成します:

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

ログイン後にコピー

このうち NewConfig() は新しい設定ファイルのインスタンスの作成に使用され、Return.Successes は各メッセージの送信が成功した場合に成功情報が返されることを示し、 NewAsyncProducer() はプロデューサーのインスタンスの作成に使用されます。 Kafka クラスター内のブローカーを表し、ノードの IP アドレスとポート番号を表します。

3. メッセージの送信:

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

ログイン後にコピー

このうち、ProducerMessage はメッセージの構造を表し、Topic はメッセージが属するトピックを表し、Value はメッセージの内容を表します。

4. メッセージコンシューマ (Consumer) インスタンスの作成:

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

ログイン後にコピー

このうち、NewConfig() は新しい設定ファイルのインスタンスを作成するために使用され、Return.Errors はメッセージが送信されるたびに消費された場合、消費失敗のエラーメッセージを返します。コンシューマインスタンスの作成には NewConsumer() が使用されます。

5. メッセージの消費:

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

ログイン後にコピー

このうち、ConsumePartition() は消費のトピック、パーティション、消費場所 (最新のメッセージまたは最も古いメッセージ) を指定するために使用され、Messages() はトピック内で消費されたメッセージから取得するために使用されます。メッセージを消費した後、MarkOffset() メソッドを使用して、メッセージが消費されたことを確認する必要があります。

Kafka リアルタイムキャッシュの実装

Golang では、Kafka メッセージキューを介してリアルタイムキャッシュを確立すると非常に便利です。プロジェクト内にキャッシュ管理モジュールを作成し、実際のニーズに応じてキャッシュの内容を対応するメッセージ構造に変換し、プロデューサーを通じて Kafka クラスター内の指定されたトピックにメッセージを送信し、コンシューマーがメッセージを消費するのを待ちます。トピックを立てて続行します。

具体的な実装手順は次のとおりです:

1. プロジェクトでキャッシュ構造とキャッシュ変数を定義します:

type Cache struct {
  Key   string
  Value interface{}
}

var cache []Cache

ログイン後にコピー

このうち、Key はキャッシュキー ( Key) 、Value はキャッシュされた値 (Value) を表します。

2. キャッシュを対応するメッセージ構造に変換します:

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

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

ログイン後にコピー

このうち、Message はメッセージ構造を表し、Operation はキャッシュ操作の種類を表し、generateMessage() はメッセージを返すために使用されます。実例。

3. プロデューサーを作成し、キャッシュされたコンテンツをメッセージとして指定されたトピックに送信します:

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
}

ログイン後にコピー

このうち、Producer() は、プロデューサーインスタンスを作成し、メッセージの受信を待機するために使用されます。パイプラインから送信されます。pushMessage() は、キャッシュされたコンテンツを Message インスタンスに変換し、プロデューサーを使用して指定されたトピックに送信するために使用されます。

4. コンシューマを作成し、指定されたトピックをリッスンし、メッセージが到着したときに対応する操作を実行します:

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

ログイン後にコピー

このうち、consumer() は、コンシューマインスタンスを作成し、コンシューマインスタンスをリッスンするために使用されます。 json.Unmarshal() 関数は、メッセージの Value フィールドを解析して Message 構造にし、Operation フィールドに基づいて対応するキャッシュ操作を実行します。メッセージを消費した後、MarkOffset() メソッドを使用して、メッセージが消費されたことを確認する必要があります。

上記の手順により、Golang で Kafka ライブラリ sarama を使用して、Kafka メッセージキューに基づくリアルタイムキャッシュテクノロジを確立することに成功しました。実際のアプリケーションでは、実際のニーズに応じてさまざまな Kafka クラスター構成とパーティションルールを選択し、さまざまなビジネスシナリオに柔軟に対応できます。

以上がGolang で Kafka メッセージキューに基づくリアルタイムキャッシュテクノロジを確立します。の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。