안녕하세요.
갈비 만두를 좋아하는 개발자 임태빈입니다.
Go에서 Channel을 활용한 간단한 파이프라인을 만들어보려고 합니다.
데이터를 생성하고 이를 Channel을 통해 전달하여 출력하고 총합을 구하는 코드를 구현해보곘습니다.
코드 구현에 앞서 그림을 간단하게 그려보면 다음과 같습니다.
그러면 이제 코드 구현을 해보겠습니다.
먼저 전체 코드를 공유하겠습니다.
package main
import "log"
func GenerateData() <-chan int {
oc := make(chan int)
go func() {
i := 0
for {
oc <- i
i = i + 1
if i == 15 {
break
}
}
close(oc)
}()
return oc
}
func PrintData(ic <-chan int) <-chan int {
oc := make(chan int)
go func() {
for data := range ic {
log.Println(data)
oc <- data
}
close(oc)
}()
return oc
}
func TotalSumOfData(ic <-chan int) {
total := 0
for data := range ic {
total += data
}
log.Println("total: ", total)
}
func main() {
TotalSumOfData(PrintData(GenerateData()))
}
첫번째로 볼 코드는 GenerateData입니다. 이 함수에서는 데이터를 생성합니다.
func GenerateData() <-chan int {
oc := make(chan int)
go func() {
i := 0
for {
oc <- i
i = i + 1
if i == 15 {
break
}
}
close(oc)
}()
return oc
}
여기서 저는 sync.WaitGroup를 사용하지 않았습니다.
그 이유는 main함수 부분에서 TotalSumOfData(PrintData(GenerateData())) 이 함수가 엮이면서 달라지기 때문입니다.
먼저 Channel을 사용하면 Channel에 값을 넣어주고 바로 소모를 시켜줘야 합니다. GenerateData에서는 그런 부분이 없지만 PrintData에서 이를 Range를 통해서 진행해줍니다. 그래서 저는 sync.WaitGroup를 사용하지 않았습니다.
두번째 코드는 PrintData입니다. 이 함수는 데이터가 잘들어왔는지를 보기 위해 작성했습니다.
func PrintData(ic <-chan int) <-chan int {
oc := make(chan int)
go func() {
for data := range ic {
log.Println(data)
oc <- data
}
close(oc)
}()
return oc
}이 부분에서는 위에서 말한 Range를 통해 GenerateData에서 생긴 Channel에 값을 소모시켜줍니다. 그런 다음 다시 TotalSumOfData를 위해 다시 새로운 Channel을 만들어서 진행합니다.
마지막으로 봐야하는 코드인 TotalSumOfData는 Channel에 있는 값들을 모두 더하여 최종값을 만들어줍니다.
func TotalSumOfData(ic <-chan int) {
total := 0
for data := range ic {
total += data
}
log.Println("total: ", total)
}저는 이 파이프라인을 만들어보면서 실시간 데이터를 처리할 수 있는 간단한 코드를 만들 수 있겠다고 생각했습니다. 다음번에는 fluentd를 통해 들어온 로그 데이터들을 처리해서 만들어보면 좋겠다는 생각이 들었습니다.
이 포스팅을 읽어주셔서 감사합니다:) 혹시나 궁금하신 부분이나 이상한 부분이 있다면 댓글 남겨주시면 감사드리겠습니다~~
