Golang의 Kafka 메시지 큐 기반 실시간 캐싱 기술을 구축합니다.

인터넷 기술의 지속적인 발전과 응용 시나리오의 지속적인 확장으로 인해 실시간 캐싱 기술은 점점 더 인터넷 회사의 필수 기술이 되었습니다. 실시간 캐싱 기술의 한 방법으로 메시지 큐는 실제 응용 프로그램의 개발자들에게 점점 더 선호되고 있습니다. 본 글에서는 Golang에서 Kafka 메시지 큐를 기반으로 실시간 캐싱 기술을 구축하는 방법을 주로 소개합니다.

Kafka 메시지 대기열이란 무엇인가요?

Kafka는 LinkedIn에서 개발한 분산 메시징 시스템으로 수천만 개의 메시지를 처리할 수 있습니다. 높은 처리량, 낮은 대기 시간, 내구성 및 높은 신뢰성의 특성을 가지고 있습니다. Kafka에는 생산자, 소비자, 주제라는 세 가지 주요 구성 요소가 있습니다. 그 중 생산자와 소비자는 Kafka의 핵심 부분입니다.

생산자는 지정된 주제에 메시지를 보내고 파티션과 키를 지정할 수도 있습니다. 소비자는 주제로부터 해당 메시지를 받습니다. Kafka에서는 생산자와 소비자가 독립적이며 서로 종속되지 않습니다. 그들은 동일한 주제를 공유함으로써만 상호 작용합니다. 이 아키텍처는 분산 메시징을 구현하고 다양한 비즈니스 시나리오에서 메시지 대기열 요구 사항을 효과적으로 해결합니다.

Golang과 Kafka의 결합

Golang은 높은 동시성, 고성능 및 기타 특성으로 인해 최근 인기 있고 효율적인 프로그래밍 언어로 점점 더 널리 사용되고 있습니다. Golang에서는 고루틴의 수가 커널 스레드의 수와 일대일 관계를 가지기 때문에 메시지 큐와 결합할 수 있다는 고유한 이점이 있습니다. 이는 Golang이 대규모 동시 작업을 효율적이고 원활하게 처리할 수 있음을 의미합니다. Kafka는 사용자 정의 가능한 파티션 규칙에 따라 다양한 메시지를 다양한 브로커 노드에 배포하여 수평적 확장을 달성할 수 있습니다.

Golang에서 타사 Kafka 라이브러리 sarama를 사용하면 Kafka와의 상호 작용을 쉽게 구현할 수 있습니다. 구체적인 구현 단계는 다음과 같습니다.

1. Golang 프로젝트에 sarama 라이브러리를 도입합니다:

import "github.com/Shopify/sarama"

2. 메시지 발신자(Producer) 인스턴스 생성:

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

그중 NewConfig()를 사용하여 새 인스턴스를 생성합니다. 구성 파일 인스턴스의 Return.Successes는 각 메시지가 성공적으로 전송되면 성공 정보가 반환된다는 것을 의미합니다. NewAsyncProducer()는 생산자 인스턴스를 만드는 데 사용됩니다. 매개 변수의 문자열 배열은 브로커 노드의 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()는 Message 인스턴스를 반환하는 데 사용됩니다.

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()는 콘텐츠를 캐시하는 데 사용됩니다. 콘텐츠는 메시지 인스턴스로 변환되고 생산자를 사용하여 지정된 주제로 전송됩니다.

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 중국어 웹사이트의 기타 관련 기사를 참조하세요!