Go에 이벤트 버스 구축하는 방법

서문

오늘날 마이크로서비스와 분산 시스템이 보편화된 환경에서 이벤트 중심 아키텍처(EDA)는 중요한 역할을 합니다. 이 아키텍처 설계는 서비스가 기존의 직접 인터페이스 호출 대신 이벤트, 동기식 또는 비동기식을 통해 통신할 수 있도록 합니다. 이벤트 기반 상호 작용 모드는 서비스 간의 느슨한 결합을 촉진하고 시스템의 확장성을 향상시킵니다.

게시-구독 패턴은 이벤트 중심 아키텍처를 구현하는 한 가지 방법입니다. 이를 통해 시스템의 다양한 구성 요소 또는 서비스가 이벤트를 게시할 수 있으며, 다른 구성 요소 또는 서비스는 이러한 이벤트를 구독하고 이벤트 내용에 따라 응답할 수 있습니다. 대부분의 개발자는 이 패턴에 익숙할 것입니다. 일반적인 기술 구현에는 메시지 큐(MQ)와 Redis의 게시/구독(PUB/SUB) 기능이 포함됩니다.

Go에서는 강력한 채널과 동시성 메커니즘을 활용하여 게시-구독 패턴을 구현할 수 있습니다. 이 기사에서는 Go에서 간단한 이벤트 버스를 구현하는 방법을 자세히 살펴보고, 이는 게시-구독 패턴의 구체적인 구현입니다.

이벤트 버스

이벤트 버스는 게시-구독 패턴의 구체적인 구현입니다. 게시자와 구독자 간의 미들웨어로서 이벤트 전달 및 배포를 관리하여 이벤트가 게시자에서 구독자로 원활하게 전송되도록 합니다.

이벤트 버스의 주요 장점은 다음과 같습니다.

디커플링: 서비스는 직접 통신할 필요 없이 이벤트를 통해 상호 작용하여 서비스 간의 종속성을 줄입니다.
비동기 처리: 이벤트를 비동기적으로 처리하여 시스템의 응답성과 성능을 향상시킬 수 있습니다.
확장성: 새로운 구독자는 기존 게시자 코드를 수정하지 않고도 이벤트에 쉽게 가입할 수 있습니다.
오류 격리: 이벤트 처리의 오류는 다른 서비스의 정상적인 작동에 직접적인 영향을 미치지 않습니다.

이벤트 버스의 코드 구현

다음으로 Go에서 간단한 이벤트 버스를 구현하는 방법을 소개합니다. 여기에는 다음과 같은 주요 기능이 포함됩니다.

게시: 시스템의 다양한 서비스가 이벤트를 보낼 수 있습니다.
구독: 관심 있는 서비스가 특정 유형의 이벤트를 구독하고 수신할 수 있습니다.
구독 취소: 서비스가 이전에 구독한 이벤트를 제거할 수 있습니다.

이벤트 데이터 구조 정의

type Event struct {
    Payload any
}

Event는 이벤트를 캡슐화하는 구조체이며, Payload는 이벤트의 컨텍스트 정보를 나타내고 그 유형은 any입니다.

이벤트 버스 정의

type (
    EventChan chan Event
)

type EventBus struct {
    mu    sync.RWMutex
    subscribers map[string][]EventChan
}

func NewEventBus() *EventBus {
    return &EventBus{
        subscribers: make(map[string][]EventChan),
    }
}

EventChan은 Event 구조체를 전달하기 위한 채널로 정의된 유형 별칭입니다: chan Event.

EventBus는 이벤트 버스의 정의입니다. 여기에는 두 가지 속성이 포함됩니다.

mu: 읽기-쓰기 뮤텍스(sync.RWMutex)로, 아래 subscribers의 동시 읽기 및 쓰기 안전성을 보장하는 데 사용됩니다.
subscribers: 키가 구독 주제를 나타내는 문자열이고 값이 EventChan 슬라이스인 맵입니다. 이 속성은 각 주제에 대한 모든 구독자를 저장하는 데 사용됩니다. 각 구독자는 자체 EventChan을 통해 이벤트를 수신합니다.

NewEventBus 함수는 새 EventBus 인스턴스를 만드는 데 사용됩니다.

이벤트 버스 메서드 구현

이벤트 버스는 이벤트 게시(Publish), 이벤트 구독(Subscribe), 이벤트 구독 취소(Unsubscribe)의 세 가지 메서드를 구현합니다.

게시

func (eb *EventBus) Publish(topic string, event Event) {
  eb.mu.RLock()
  defer eb.mu.RUnlock()
  // Copy a new subscriber list to avoid modifying the list while publishing
  subscribers := append([]EventChan{}, eb.subscribers[topic]...)
  go func() {
    for _, subscriber := range subscribers {
      subscriber <- event
    }
  }()
}

Publish 메서드는 이벤트를 게시하는 데 사용됩니다. 이 메서드는 두 개의 매개변수, topic(주제)과 event(캡슐화된 이벤트 객체)를 수신합니다.

Publish 구현에서 읽기 잠금은 먼저 mu 속성을 통해 획득되어 subscribers에 대한 다음 작업이 동시 루틴에서 안전한지 확인합니다. 그런 다음 주제에 대한 현재 구독자 목록의 복사본이 만들어집니다. 새 goroutine이 시작되고, 복사된 구독자 목록을 반복하고 각 구독자의 채널을 통해 이벤트를 보냅니다. 이러한 작업이 완료되면 읽기 잠금이 해제됩니다.

구독자 목록의 복사본을 만드는 이유는 무엇입니까?

답변: 구독자 목록을 복사하면 이벤트를 보내는 동안 데이터 일관성과 안정성이 보장됩니다. 채널로 데이터를 보내는 것은 새 goroutine에서 수행되기 때문에 데이터를 보낼 때 읽기 잠금이 이미 해제되었을 수 있으며 구독자를 추가하거나 제거하여 원래 구독자 목록이 변경되었을 수 있습니다. 원래 구독자 목록을 직접 사용하면 예기치 않은 오류가 발생할 수 있습니다(예: 닫힌 채널로 데이터를 보내면 패닉이 발생할 수 있음).

func (eb *EventBus) Subscribe(topic string) EventChan {
  eb.mu.Lock()
  defer eb.mu.Unlock()
  ch := make(EventChan)
  eb.subscribers[topic] = append(eb.subscribers[topic], ch)
  return ch
}

Subscribe 메서드는 특정 주제에 대한 이벤트를 구독하는 데 사용됩니다. 구독할 주제를 지정하는 topic 매개변수를 허용합니다. 이 메서드를 통해 주제에 대한 이벤트를 수신할 수 있는 EventChan 채널을 얻을 수 있습니다.

Subscribe 구현에서 쓰기 잠금은 먼저 mu 속성을 통해 획득되어 subscribers에 대한 다음 읽기 및 쓰기 작업이 동시 루틴에서 안전한지 확인합니다. 그런 다음 새 EventChan 채널 ch가 생성되어 관련 주제의 구독자 슬라이스에 추가됩니다. 이러한 작업이 완료되면 쓰기 잠금이 해제됩니다.

구독 취소

func (eb *EventBus) Unsubscribe(topic string, ch EventChan) {
  eb.mu.Lock()
  defer eb.mu.Unlock()
  if subscribers, ok := eb.subscribers[topic]; ok {
    for i, subscriber := range subscribers {
      if ch == subscriber {
        eb.subscribers[topic] = append(subscribers[:i], subscribers[i+1:]...)
        close(ch)
        // Drain the channel
        for range ch {
        }
        return
      }
    }
  }
}

Unsubscribe 메서드는 이벤트 구독을 취소하는 데 사용됩니다. 두 개의 매개변수, topic(구독한 주제)과 ch(발행된 채널)를 수신합니다.

Unsubscribe 메서드 내부에서 쓰기 잠금은 먼저 mu 속성을 통해 획득되어 subscribers에 대한 다음 작업에 대한 동시 읽기 및 쓰기 안전성을 보장합니다. 그런 다음 주제에 해당하는 구독자가 있는지 확인합니다. 있는 경우 해당 주제에 대한 구독자 슬라이스를 순회하고 ch와 일치하는 채널을 찾아 구독자 슬라이스에서 제거하고 채널을 닫습니다. 그런 다음 채널이 비워집니다. 이러한 작업 후 쓰기 잠금이 해제됩니다.

사용 예

package main

import (
  "fmt"
  "time"

  "demo-eventbus"
)

func main() {
  eventBus := eventbus.NewEventBus()

  // Subscribe to the "post" topic event
  subscribe := eventBus.Subscribe("post")

  go func() {
    for event := range subscribe {
      fmt.Println(event.Payload)
    }
  }()

  eventBus.Publish("post", eventbus.Event{Payload: map[string]any{
    "postId": 1,
    "title":  "Welcome to Leapcell",
    "author": "Leapcell",
  }})
  // Topic with no subscribers
  eventBus.Publish("pay", eventbus.Event{Payload: "pay"})

  time.Sleep(time.Second * 2)
  // Unsubscribe from the "post" topic event
  eventBus.Unsubscribe("post", subscribe)
}

확장을 위한 제안

이 기사에서 구현된 이벤트 버스는 비교적 간단합니다. 이벤트 버스의 유연성, 안정성 및 유용성을 향상시키려면 다음과 같은 방법으로 확장을 고려할 수 있습니다.

이벤트 지속성: 시스템 충돌 후 처리되지 않은 이벤트를 복구할 수 있도록 이벤트에 대한 영구 스토리지를 구현합니다.
와일드카드 및 패턴 일치 구독: 단일 특정 주제가 아닌 관련 주제 그룹을 구독하기 위해 와일드카드 또는 정규식을 사용할 수 있습니다.
로드 밸런싱 및 메시지 배포 전략: 로드 밸런싱을 달성하기 위해 여러 구독자 간에 이벤트를 배포합니다.
플러그인 지원: 로깅, 메시지 필터링, 변환 등과 같은 플러그인을 통해 기능 확장을 활성화합니다.

결론

이 기사에서는 Go에서 간단한 이벤트 버스를 구현하는 프로세스를 자세히 살펴봅니다. 채널 및 동시성 메커니즘과 같은 Go의 강력한 기능을 활용하여 게시-구독 패턴을 쉽게 구현할 수 있습니다.

이 기사는 디커플링, 비동기 처리, 확장성 및 오류 격리를 포함한 이벤트 버스의 장점을 소개하는 것으로 시작합니다. 그런 다음 이벤트 데이터 구조 및 이벤트 버스 구조를 정의하는 방법과 이벤트 게시, 구독 및 구독 취소에 대한 메서드를 구현하는 방법을 자세히 설명합니다. 마지막으로 이벤트 지속성, 와일드카드 구독, 로드 밸런싱 및 플러그인 지원과 같은 몇 가지 잠재적인 확장 방향을 제안하여 이벤트 버스의 유연성과 기능을 향상시킵니다.

이 기사를 읽음으로써 Go에서 간단하면서도 강력한 이벤트 버스를 구현하고 가능한 요구 사항에 따라 확장하는 방법을 배울 수 있습니다.