[Go] 고루틴과 채널을 활용한 이벤트 처리 시스템

고루틴과 채널을 활용한 이벤트 처리 시스템

Go 언어는 동시성(concurrency)을 간단하고 효율적으로 처리할 수 있는 강력한 기능인 고루틴(goroutines)과 채널(channels)을 제공합니다. 이들을 이해하고 올바르게 사용하는 것은 Go에서 효과적인 프로그래밍을 위해 필수적입니다. 이 글에서는 고루틴과 채널의 개념을 쉽게 설명하고, 블로킹(blocking)과 넌블로킹(non-blocking)의 차이를 설명한 후, 실제 예제 코드를 통해 설명을 더욱 구체화합니다.

고루틴과 채널의 개념

고루틴(Goroutines)

고루틴은 Go 런타임에 의해 관리되는 경량 스레드(lightweight thread)입니다. 고루틴을 사용하면 함수나 메서드를 동시에 실행할 수 있습니다. 고루틴은 go 키워드를 함수 호출 앞에 붙여 간단하게 생성할 수 있으며, Go 런타임은 필요에 따라 이러한 고루틴들을 스케줄링합니다.

채널(Channels)

채널은 고루틴 간에 데이터를 전송하고 동기화하는 데 사용되는 파이프입니다. 채널을 통해 한 고루틴이 다른 고루틴에게 값이나 객체를 안전하게 전달할 수 있습니다. 채널은 생성 시점에 타입을 지정받으며, 해당 타입의 값만을 전송할 수 있습니다.

블로킹(Block)과 넌블로킹(Non-Block)

블로킹

블로킹은 특정 조건이 만족될 때까지 실행을 멈추고 기다리는 것을 의미합니다. 예를 들어, 채널에서 데이터를 읽을 때, 데이터가 도착할 때까지 고루틴의 실행이 블로킹되어 대기할 수 있습니다.

넌블로킹

넌블로킹은 실행 중인 작업이 즉시 완료될 수 없을 때 멈추지 않고 계속 진행하는 것을 의미합니다. 채널에서 데이터를 읽을 때, 데이터가 없으면 다른 작업으로 넘어가며, 이는 select 문과 default 케이스를 통해 구현할 수 있습니다.

 

버퍼링된 채널과 버퍼링되지 않은 채널의 개념

버퍼링되지 않은 채널

package main

import (
	"fmt"
	"time"
)

func main() {
	ch := make(chan int) // Unbuffered channel

	go func() {
		fmt.Println("Sending value...")
		ch <- 42 // Blocks until the value is received
		fmt.Println("Value sent")
	}()

	time.Sleep(time.Second) // Simulate work to show the blocking behavior
	fmt.Println("Receiving value...")
	val := <-ch // Blocks until a value is sent
	fmt.Println("Value received:", val)
}

버퍼링되지 않은 채널에는 값을 저장할 수 있는 용량이 없습니다. 버퍼링되지 않은 채널로 값이 전송되면 다른 고루틴이 값을 수신할 준비가 될 때까지 전송 작업이 차단됩니다. 마찬가지로, 버퍼링되지 않은 채널에서 수신하면 값이 전송될 때까지 수신 고루틴이 차단됩니다. 이러한 직접적인 핸드오프를 통해 발신자와 수신자가 서로 정확하게 동기화할 수 있으므로 버퍼링되지 않은 채널이 고루틴 조정에 이상적입니다.

이 예에서 채널에 값을 보내는 고루틴은 기본 고루틴이 값을 수신할 때까지 차단됩니다. 'time.Sleep' 호출은 차단 동작을 보여줍니다.

 

버퍼링된 채널

package main

import (
	"fmt"
)

func main() {
	ch := make(chan int, 2) // Buffered channel with a capacity of 2

	ch <- 10 // Non-blocking send
	ch <- 20 // Non-blocking send

	fmt.Println("Sending values without immediate receivers")

	val := <-ch // Non-blocking receive
	fmt.Println("Received value:", val)

	val = <-ch // Non-blocking receive
	fmt.Println("Received value:", val)
}

버퍼링된 채널에는 전송 작업을 차단하기 전에 하나 이상의 값을 저장할 수 있는 용량이 있습니다. 버퍼링된 채널의 수신 작업은 채널이 비어 있지 않은 한 차단되지 않습니다. 버퍼링된 채널은 전송 및 수신 작업을 분리하여 즉각적인 동기화 없이 goroutine이 계속 실행될 수 있도록 합니다.

여기서 'ch' 채널의 용량은 2이므로 차단 없이 두 개의 값을 전송할 수 있습니다. 채널의 버퍼가 가득 찬 경우에만 값 블록을 보냅니다. 마찬가지로 채널에서 수신하는 것은 비어 있지 않은 한 차단되지 않습니다.

 

버퍼링된 채널과 버퍼링되지 않은 채널 중에서 선택

버퍼링되지 않은 채널: 고루틴을 동기화하려는 경우 버퍼링되지 않은 채널을 사용하여 진행하기 전에 정확히 하나의 고루틴에서 전송이 수신되는지 확인하세요.

버퍼 채널: 전송된 값의 즉각적인 수신을 기다리지 않고 고루틴이 진행되도록 허용해야 하는 경우 버퍼 채널을 사용합니다. 이는 동시에 실행되는 고루틴 수를 제한하거나 리소스 풀을 관리하려는 경우에 도움이 될 수 있습니다.

고루틴과 채널을 활용한 이벤트 처리 시스템 예제 코드 

package event_handler

import (
	"fmt"
	"github.com/google/uuid"
	"testing"
	"time"
)

type Event interface {
	Process()
}

type AtomicEvent struct {
	ID        string
	Msg       string
	CreatedAt time.Time
}

type EventA struct {
	AtomicEvent
}

func (e EventA) Process() {
	time.Sleep(time.Millisecond * 500)
	fmt.Println("[event-received] event processing...", e.Msg, e.ID)
}

type EventB struct {
	AtomicEvent
}

func (e EventB) Process() {
	time.Sleep(time.Millisecond * 500)
	fmt.Println("[event-received] event proccessing..", e.Msg, e.ID)
}

func TestEventHandler(t *testing.T) {

	t.Run("event-handler", func(t *testing.T) {

		// Unbuffered channel
		//eventChannel := make(chan Event) 

		// Buffered channel
		eventChannel := make(chan Event, 15)

		// Blocking event channel
		//eventConsumer(eventChannel)
        
		// Non-Blocking event channel
		eventConsumer2(eventChannel)
        
		eventProducer(eventChannel)
	})
}

func eventProducer(eventChannel chan Event) {
	ticker := time.NewTicker(time.Millisecond * 50)
	defer ticker.Stop()

	timeout := time.After(time.Minute * 1)

	for {
		select {

		case <-ticker.C:
			if time.Now().Unix()%2 == 0 {
				sendEventA(eventChannel)
			} else {
				sendEventB(eventChannel)
			}

		case <-timeout:
			fmt.Println("[event] terminating...")
			close(eventChannel)
			return
		}
	}
}

func sendEventA(eventChannel chan Event) {
	id := uuid.NewString()
	eventChannel <- EventA{
		AtomicEvent{
			ID:        id,
			Msg:       "This is EventA",
			CreatedAt: time.Now(),
		},
	}
	fmt.Println("[event] Sent EventA. id: ", id)
}
func sendEventB(eventChannel chan Event) {
	id := uuid.NewString()
	eventChannel <- EventB{
		AtomicEvent{
			ID:        id,
			Msg:       "This is EventB",
			CreatedAt: time.Now(),
		},
	}
	fmt.Println("[event] Sent EventB. id: ", id)
}

func eventConsumer(eventChannel chan Event) {
	go func() {
		for event := range eventChannel {
			event.Process()
		}
	}()
}

func eventConsumer2(eventChannel chan Event) {
	go func() {
		for {
			select {
			case event := <-eventChannel:
				event.Process()
			default:
				fmt.Println("not found event channel")
			}
		}
	}()
}

예제 코드 설명

위에서 제공된 코드는 이벤트 처리 시스템의 구현 예시입니다.
이 시스템은 Event 인터페이스와 두 가지 구체적인 이벤트 타입(EventA, EventB)을 정의하고, 이벤트 생산자(eventProducer)와 소비자(eventConsumer, eventConsumer2)를 통해 이벤트를 처리합니다.

eventProducer는 주기적으로 이벤트를 생성하여 채널에 전송하고, eventConsumer 또는 eventConsumer2가 이벤트를 받아 처리합니다. 이 과정에서 고루틴은 비동기적으로 동작하며, 채널은 이벤트 전송을 위한 동기화 메커니즘 역할을 합니다.

eventProducer에서는 ticker와 timeout을 이용해 이벤트 생성을 주기적으로 관리하고, 일정 시간 후에 이벤트 생성을 종료합니다.
이러한 패턴은 타임아웃이나 주기적인 작업 처리에 유용하게 활용할 수 있습니다.

 

eventConsumer()와 eventConsumer2()의 동작 방식 비교

• eventConsumer() 함수는 채널에서 이벤트를 읽을 때까지 블로킹되며, 이벤트가 도착하면 해당 이벤트를 처리합니다.
  이 방식은 CPU 자원을 효율적으로 사용합니다.
• 반면, eventConsumer2() 함수는 select 문과 default 케이스를 사용하여 채널을 넌블로킹으로 폴링합니다.
  이벤트가 없을 경우 "not found event channel" 메시지를 출력하고 계속해서 루프를 실행합니다. 이 방식은 채널에 이벤트가 없을 때도 CPU 자원을 소모하게 만들며, 이벤트가 드물게 발생하는 시스템에서는 효율적이지 않을 수 있습니다.

eventConsumer() 함수는 채널에서 이벤트를 기다리는 동안 다른 작업으로 CPU 자원을 전환할 수 있어 자원을 절약할 수 있습니다. 이벤트가 도착하면 즉시 이벤트 처리를 시작합니다. 반면, eventConsumer2()는 계속해서 채널 상태를 확인하기 때문에, 이벤트 대기 중에도 CPU를 사용합니다. 이는 특히 멀티코어 시스템에서 비효율적일 수 있으며, 불필요한 자원 사용을 초래할 수 있습니다.

 

결론

고루틴과 채널은 Go의 강력한 동시성 관리 도구입니다. 이들을 사용하면 다양한 동시성 패턴을 간단하고 효율적으로 구현할 수 있습니다. 블로킹과 넌블로킹 방식의 이해는 이러한 도구를 보다 효과적으로 사용하기 위해 중요합니다. 위 예제 코드는 이벤트 처리 시스템을 구현하는 한 방법을 보여주며, 고루틴과 채널의 사용 방법에 대한 이해를 돕습니다. 실제 애플리케이션 개발에서는 이러한 패턴을 참고하여 효율적이고 안정적인 동시성 처리 로직을 설계할 수 있습니다.