goroutine과 channel로 알아보는 비동기 처리

서문

안녕하세요?

 

오늘은 go언어에서 channel과 goroutine에 대해 이야기해보려고 합니다.

goroutine에 대하여

goroutine은 가상 스레드로 처리를 시작하는 기능입니다.

 

작업자를 늘려서 일을 분담한다는 의미와 비슷합니다.

 

아래 코드와 같이 go 키워드를 사용하여 함수를 호출하면 goroutine으로 시작됩니다.

func main() {
    go task()
}

func task() {
    // 어떤 작업
}

 

여기서 goroutine으로 시작하고 싶은 함수 즉 task 함수에 반환값을 설정하고 싶은 경우가 있는데요.

 

그냥 쉽게 생각하면 다음과 같은 코드를 작성할 수 있습니다.

func main() {
    // 문법 오류가 발생합니다
    // 원래는 반환값을 지정할 수 없지만, 만약 가능하다고 가정하면
    s := go task()
}

func task() string {
    // 어떤 작업
    return "작업 완료."
}

 

이렇게 반환값을 받는 스타일로 하면,

 

s := go task() 부분에서 s를 받기 위해 task의 결과를 기다려야 하므로,

 

결론적으로 goroutine의 의미인 비동기로 처리할 수 없습니다.

 

그래서 goroutine으로 시작하고 싶은 함수에는 반환값을 설정할 수 없도록 되어 있습니다.

 

또한, goroutine은 처리가 완료되면 파기됩니다.

 

GoLang에서 완료된 처리에 대해 무리하게 스레드를 유지하는 일은 없습니다.

 

매번 파기하고, 리소스를 열어줍니다.

 

goroutine을 늘리는 것은 간단합니다.

 

go 키워드로 함수를 많이 호출해주기만 하면 됩니다.

func main() {
    // 10번 비동기로 task를 호출합니다
    for i := 0; i < 10; i++ {
        go task()
    }
}

func task() {
    // 어떤 작업
}

 

이렇게 하면 병렬로 작업 처리가 수행됩니다.

---

프로세스 간 정보의 불투명성

지금까지 살펴본 바로는 goroutine이 비동기 처리를 잘 해내어 기계 자원을 효율적으로 사용할 수 있을 것 같다는 것을 알 수 있었습니다.

 

하지만, 이것만으로는 문제가 있습니다.

 

goroutine에는 다음과 같은 특징이 있습니다.

• 반환값이 없음
• 처리가 완료되면 파기됨

이대로라면 goroutine이 '작업 중인지, 파기되었는지'를 알 수 있는 방법이 없습니다.

 

또한, goroutine이 언제 끝나는지(또는 이미 끝났는지)를 알 수 없기 때문에, goroutine의 작업을 기다릴 수도 없습니다.

 

예를 들어, 앞서 10번 goroutine을 실행하는 코드를 다음과 같이 업데이트해보면 이를 쉽게 알 수 있습니다.

func main() {
    // 10번 비동기로 task를 호출합니다
    for i := 0; i < 10; i++ {
        go task()
    }
}

func task() {
    // 어떤 작업
    fmt.Println("I'm goroutine.")
}

 

이 코드를 실행해보면, (아마도 대부분의 환경에서) 'I'm goroutine.'이 한 번도 출력되지 않을 것입니다.

 

이는 main 스레드가 task의 처리를 기다릴 수 없기 때문에, for문을 돌며 goroutine을 실행하기만 하고 바로 종료되기 때문입니다.

 

또한, 어떤 실수로 goroutine의 작업이 멈추었다고 해도, 그것을 감지할 수는 없습니다.

func main() {
    rand.Seed(time.Now().UnixNano())
    for i := 0; i < 10; i++ {
        // 0 또는 1을 랜덤으로 생성합니다
        go task(rand.Intn(2))
    }
}

func task(random int) {
    // 어떤 작업
    switch random {
    case 0:
        // 실패합니다
    case 1:
        // 성공합니다
    }
}

 

감지할 수 없다는 것은, 의도하지 않은 동작이 발생했을 때 에러 핸들링이 불가능하다는 것을 의미합니다.

 

또한, 반환값이 없기 때문에 작업 결과로서 가공된 값을 받을 수도 없습니다.

 

이러한 점들로 인해, 이대로라면 goroutine의 처리는 반환값을 필요로 하지 않는(즉, 에러 핸들링이 필요 없는) 것만 가능하게 됩니다.

 

그렇게 되면 매우 사용하기 어려울 것 같죠.

 

하지만 GoLang에는 'channel'이라는 기능이 제공되고 있습니다.

---

channel에 대하여

channel은 프로세스 간에 값을 주고받기 위한 기능입니다.

 

채널을 사용하면 다음과 같은 것들이 가능해집니다.

• goroutine과의 값을 주고받기
• '서로 지나치지 않기' 위한 대기

채널은 goroutine 간의 편지와 같은 것으로, goroutine을 넘나들며 값을 쓰고 받을 수 있습니다.

 

일단 채널의 값 쓰기는 다음과 같이 합니다.

ch <- val

 

그리고 값 받기는 다음과 같이 합니다.

val <- ch

 

다른 언어를 공부하다가 GoLang으로 오시면 처음에는 이 화살표가 조금 이해하기 어렵운데요.

 

여기서 한번 정리해보죠.

 

채널을 편지에 비유한다면, 오른쪽에서 오는 것이 송신이고 왼쪽으로 나가는 것이 수신입니다.

 

채널의 수신을 코드로 작성하면, 채널은 값이 송신될 때까지 기다려줍니다.

 

이렇게 하면 서로 지나치지 않고 코드를 작성할 수 있습니다.

func main() {
    // 채널 생성
    txtCh := make(chan string)

    // 사용 후 닫기
    defer close(txtCh)

    // goroutine에 채널 전달
    go task(txtCh)

    // 채널 수신
    // 이 때, 할당도 가능
    s := <-txtCh
    fmt.Println(s)
}

// 채널을 쓰기 전용으로 받기
func task(txtCh chan<- string) {
    // 채널에 결과 쓰기
    txtCh <- "I'm goroutine"
}

 

채널은 버퍼를 가질 수도 있습니다.

 

생성 시 버퍼를 지정하면, 채널이 가질 수 있는 데이터의 개수가 결정됩니다.

 

기본값은 0(직접 전달해야 하며, 저장할 수 없음)입니다.

 

버퍼를 초과하여 값을 송신하면, 송신 블로킹이 발생합니다.

func main() {
    // 채널 생성
    // 기본 버퍼는 0(직접 전달)
    txtCh := make(chan string)
    defer close(txtCh)

    // task 실행
    go task(txtCh)

    // 값 송신
    txtCh <- "push txt1."

    // 수신되기 전에 다시 송신
    // 송신 블로킹 발생
    // 이후 수신자가 없어 무한히 기다리게 되므로 데드락 발생
    txtCh <- "push txt2."
}

// 채널을 수신 전용으로 받기
func task(txtCh <-chan string) {
    // 1초 기다림
    time.Sleep(1 * time.Second)
    // 채널에서 값 받기
    <-txtCh
}

 

위 코드의 실행결과는 주석에서 설명했듯이 데드락이 발생합니다.

go run main.go
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
        /Users/Codings/Golang/blog/go-channel/main.go:20 +0xb8
exit status 2

 

그리고 GoLang에서 버퍼를 설정하면, 그만큼의 값을 보관할 수 있습니다.

func main() {
    // 채널 생성
    // 버퍼를 10까지 설정
    txtCh := make(chan string, 10)
    defer close(txtCh)

    // task 실행
    go task(txtCh)

    // txtCh에 값 10개 송신
    // 버퍼에 들어가므로 블로킹이 발생하지 않음
    for i := 0; i < 10; i++ {
        txtCh <- fmt.Sprintf("push txt%d.", i)
    }

    // 채널로의 송신이 블록되지 않으므로
    // goroutine의 결과를 기다리지 않고 처리가 종료됨
}

// 채널을 수신 전용으로 받기
func task(txtCh <-chan string) {
    // 1초 기다림
    time.Sleep(1 * time.Second)
    // 채널에서 값 받기
    <-txtCh
}

 

위 코드를 실행하면 에러없이 잘 실행됩니다.

 

송신의 블로킹과 버퍼를 이용하여, '특정 범위 접근 처리의 워커 수를 지정하는' 것과 같은 일도 할 수 있습니다.

func main() {
    // 버퍼 6으로 스레드 수를 관리하는 채널 생성
    limitCh := make(chan struct{}, 6)
    defer close(limitCh)

    // 결과를 받는 채널 생성
    resCh := make(chan string)
    defer close(resCh)

    // 100개의 요소를 가진 slice
    s := []int{
        100,
        200,
        300,
        400,
        // ...생략
    }

    // 요소 수 기억
    count := 0
    for _, n := range s {
        // task가 내부에서 limitCh의 송수신을 함
        // 6번 루프하면 버퍼가 차서 송신 블로킹 발생
        // goroutine이 해제되면 limitCh의 버퍼에 여유가 생겨 송신 블로킹이 해제됨
        // 즉, 동시에 6개까지만 goroutine이 실행될 수 있음
        go task(n, limitCh, resCh)
        count++
    }

    // 꺼내기
    // resCh와 count를 세트로 반환하는 함수로 만들어, 다른 곳에서 꺼내기도 가능
    for i := 0; i < count; i++ {
        fmt.Println(<-resCh)
    }
}

func task(n int, limitCh chan struct{}, resCh chan<- string) {
    // limitCh를 송신(버퍼 1개 확보)
    limitCh <- struct{}{}
    // 작업 수행
    resCh <- fmt.Sprintf("message %d", n)
    // 리미트 수신(버퍼 1개 해제)
    <-limitCh
}

---

channel + select

channel은 select를 사용하여 유연하게 송수신할 수 있습니다.

 

기본적으로는 다음과 같이 작성할 수 있습니다.

select {
case <-ch1:
    // ch1의 경우
}

 

여러 패턴도 받을 수 있습니다.

 

select는 어떤 조건에 맞을 때까지 처리되지 않습니다.

 

만약 어느 한 조건에 맞으면 select에서 빠져나옵니다.

select {
case <-ch1:
    // ch1의 경우
case <-ch2:
    // ch2의 경우
}

 

즉, case로 여러 채널을 수신하면, 어느 하나의 채널을 받을 때까지 처리가 블로킹됩니다.

 

그리고 값의 할당도 가능합니다.

select {
case s := <-ch1:
    fmt.Println(s)
case n := <-ch2:
    fmt.Println(n)
}

 

쓰기에도 select를 사용할 수 있습니다.

 

쓰기의 경우는 가장 먼저 버퍼가 비어 있는 채널에 값이 전송되고 select를 빠져나옵니다.

select {
case ch1 <- s:
    // ch1의 경우
case ch2 <- s:
    // ch2의 경우
}

 

블로킹하고 싶지 않은 경우, default를 설정해줍니다.

 

default는 어떤 조건에도 맞으므로, 어떤 타이밍에서도 통과합니다.

 

무한 루프 등과 결합하여 사용하는 경우가 많습니다.

// 기본적으로는 아무것도 하지 않고, ch1 채널에 값이 올 때까지 계속 기다립니다.
for {
    select {
    case ch1 <- 1:
        // ch1의 경우
        // ch1을 받았을 때만 아래 동작을 수행합니다.
    default:
        // 기본 패턴
    }
}

표준 패키지로의 goroutine 관리

goroutine을 관리할 수 있는 패키지로는 다음과 같은 것들이 있습니다.

• context: 취소 처리나 값의 전파
• sync: 비동기 처리에 필요한 다양한 기능을 제공
• sync/atomic: 저수준 동기 처리를 위한 패키지. 기본적으로 channel과 sync로 충분하다고 문서에 있음

이러한 패키지들은 goroutine 간의 취소 신호를 관리해주거나, 공통으로 다루는 값의 경쟁을 관리해줍니다.

 

이 글에서는 자세히 다루지 않겠지만, 관심 있는 분은 꼭 알아보시길 바랍니다.

 

이처럼 GoLang에는 channel이라는 기능을 중심으로, goroutine을 더 잘 관리할 수 있게 해주는 기능을 제공하고 있습니다.

마지막으로

GoLang에서 가장 GoLang다운 처리는 비동기 처리인데요.

 

그래서 GoLang으로 웹 애플리케이션을 많이 만듭니다.

 

비동기 처리에서 메시징을 잘 할 수 있도록, channel이라는 기능을 제공해 줍니다.

 

Golang에서 channel은 매우 잘 만들어져 있어, 다양한 비동기 패턴을 구현할 수 있습니다.

 

그리고 그 비동기 패턴을 잘 구현하기 위한 도구가 표준 패키지로 제공되고 있어 프로그래머 입장에서는 사용하기 아주 편한데요.

 

이러한 충실한 지원 덕분에, GoLang으로 비동기 처리를 하는게 다른 언어보다 훨씬 쉬운거 같습니다.

---