빠르게 훝어 보는 node.js - async 프레임웍을 이용한 콜백헬의 해결

빠르게 훝어 보는 node.js - async 프레임웍을 이용한 콜백헬의 해결

조대협 (http://bcho.tistory.com)

콜백헬의 정의

 

node.js는 자바스크립트의 콜백 패턴을 사용한다. 그래서 함수들을 순차적으로 실행하고자 할때 콜백 함수들의 중첩이 생겨서 코드가 복잡해지는 문제가 생긴다. 코드가 복잡해지고, 코드의 가독성이 떨어져서 유지 보수가 매우 힘들어진다.

 

파일을 읽어서 쓰는 코드를 보자

var fs = require('fs'); var src = '/tmp/myfile.txt'; var des = '/tmp/myfile_async.txt';   fs.readFile(src,'utf-8',function(err,data){             console.log(data);       if(err){            console.log("Read file error");       }else{            console.log("Read file is done");            fs.writeFile(des,data,function(err){                  if(err){                        console.log("Write file error");                        return;                  }                  console.log("Write file is done");            });       } });

Figure 1 파일을 읽어서 쓰는 코드에서 콜백이 중첩된 예제

 

파일을 읽은후에, 파일을 쓰려면, 파일을 읽는 함수 readFile에서 파일을 다 읽은 후에 호출되는 콜백 함수에서 writeFile 함수를 호출해야 한다.

만약에 위의 예제처럼 두개의 비동기 함수가 아니라, 여러개의 비동기 함수를 순차적으로 실행해야 한다면?

아래코드를 보면 6개의 비동기 함수를 순차적으로 호출하기 위한 코드인데, 콜백 함수가 6번 중첩이 되었음을 볼 수 있다.

알고리즘을 제외한 코드인데, 알고리즘이 들어가 있다면 코드는 훨씬 복잡해지게 된다.

 

asyncfunction(params,function(){       asyncfunction(params,function(){            asyncfunction(params,function(){                  asyncfunction(params,function(){                        asyncfunction(params,function(){                              asyncfunction(params,function(){                              });                        });                  });            });       }); });

Figure 2 callback hell의 개념

 

이러한 복잡성을 해결해주기 위해서, 자바스크립트에서는 몇몇 프레임웍을 제공하는데 대표적으로 사용되는 프레임웍으로는 async (https://github.com/caolan/async) 와 promise.js (https://www.promisejs.org/) 가 있다.

Async

async는node.js의 콜백헬 문제를 풀기 위해서 개발되었지만, 현재는 브라우져에서도 사용이 가능하며 자바스크립 기반의 애플리케이션의 콜백헬 문제를 푸는데도 사용이 가능하다. 콜백헬 뿐 아니라, 20여가지의 추가 함수를 지원하고 있고, parallel과 같은 동시 수행이 가능한 코드의 동시성 제어로도 다양하게 사용이 가능하다.

여기서는 async에서 자주 사용되는 동시성 제어 흐름에 대해서 알아보도록 한다.

 

waterfall

waterfall은 흐름제어에 있어서 여러개의 비동기 함수를 순차적으로 실행하되, 앞의 비동기 함수의 결과 값을 뒤의 비동기 함수에 인자로 전달하는 흐름이다.

 

Figure 3 waterfall 흐름 제어의 개념

이 그림은 비동기 함수 asyncfunctionaA, asyncfunctionB,asyncfunctionC 를 순차적으로 실행하고, 각 단계에서 다온 리턴값을 다음 단계로 넘기는 waterfall 흐름의 개념을 표현하고 있다. 각각의 단계에서 처리되는 함수를 async에서는 task라고 정의한다.

이 task가 모두 수행이 끝나면, 맨 마지막에 정의된 callback 함수가 수행된다.

만약task 수행도중에 에러가 발생하면, task 수행을 멈추고 callback 함수를 바로 호출하는데, 이때 err라는 인자에 에러 내용을 채워서 넘긴다. 맨 마지막 callback함수는 err에 null이면 정상적으로 모든 task들이 성공적으로 호출된것으로 처리하고, 만약에 null이 아닌경우 task 수행도중에 에러가 난것으로 파악하여 에러 처리를 한다.

 

waterfall의 문법을 살펴보자 

waterfall(tasks,[callback])

Figure 4 waterfall 제어 흐름 문법

waterfall에는 두가지 인자를 넘기도록 되어 있다.

·         첫번째 인자 tasks는 배열로, 순차적으로 실행될 함수들을 배열로 정의한다.

·         두번째 인자는 callbac(err,[result])으로, 모든 함수가 순차적으로 끝난후에 맨 마지막에 수행되는 함수이다. 또한 tasks를 실행하다가 에러가 발생하면 이 최종 callback을 호출한다.
task를 실행하다가 에러가 발생하면 실행을 멈추고 이 최종callback으로 첫번째 인자인 err에 에러 내용을 넣어서 전달한다., 만약에 에러가 발생하지 않았을 경우에는 모든 task 완료한 후에 err에 ‘null’을 전달한다. 두번째 인자는 생략이 가능한데, 마지막 tasks에서 넘어온 결과에 대한 값을 저장하고 있는 변수 이다.
선택적으로 result 인자를 정의할 수 있는데, 이 경우 waterfall 에 정의된 task의 맨마지막 task (최종callback 이전에 바로 실행된 task)의 리턴값을 넘겨 받는다.

 

이해를 돕기 위해서 코드를 보자. 아래 코드는 위의 그림에 표현된 asyncfunctionA,B,C를 순차적으로 호출하는 흐름을 waterfall로 표현한 슈도 코드이다. (개념을 돕기위한 코드로 실제로 실행이 되지는 않는다).

var async = require('async');   async.waterfall([               function(callback){                  asyncfunctionA(param,callback);               },               function(resultA,callback){                  asyncfunctionB(resultA,callback);               },               function(resultB,callback){                  asyncfunctionC(resultB,callback);               }              ],              function(err,resultC){                        if(err) errorHandler(err);                              // handle resultC                   } );

Figure 5 waterfall 제어 흐름의 사용 방법 (psedo code)

 

waterfall함수에 첫번째 인자는 배열 형태로 function(callback), function(resultA,callback),function(resultB,callback)을 기술하였다. 각 함수에서는 우리가 호출할 비동기 함수 asyncfunctionA,B,C를 각각 호출하였다.

배열에 인자로 들어가 있는 각 함수는 맨마지막 인자로 callback을 전달 받는데, 이 callback은 다음 호출한 함수를 지칭한다.

맨 처음 호출한 function(callback)에서 이 callback은 function(resultA,callback)을 지칭하고, 여기에 있는 callback은 다음  function(resultB,callback)을 지칭한다.

 

function(callback){                  asyncfunctionA(param,callback);               }

 

에서 asyncfunctionA에서 callback을 인자로 넘겼는데, 원래 asyncfunctionA의 함수 정의가 다음과 같다.

asyncfunctionA = function(param,function(resultA){ …                  }

 

asyncfunctioA는 비동기 함수로 실행이 끝나면 callback함수를 호출하게 되어 있는데, 이 callback함수의 인자는 resultA를 받게되는 있는 형태이다.

그런 이유로waterfall 로 넘겨지는 함수 배열중 두번째 함수의 형이 function(resultA,callback) 형태를 띄게 되는 것이다.

 

실제로 작동하는 코드를 구현해 보자. 아래 코드는 앞서 async없이 작성했던 파일을 읽어서 다른 파일에 쓰는 코드를 async의 waterfall을 이용하여 구현한 예이다.

 

아래 예제를 실행하기 위해서는 package.json에 “async” 의존성을 추가하거나, 또는 실행 환경에서

%npm install async

를 실행해서 async 모듈을 설치해야 한다.

 

var async = require('async');   var fs = require('fs'); var src = '/tmp/myfile.txt'; var des = '/tmp2/myfile_async.txt';   async.waterfall([               function(callback){                  fs.readFile(src,callback);               },               function(data,callback){                  fs.writeFile(des,data,callback);               }              ],              function(err){                        if(err) console.log(err);                   } );

Figure 6 waterfall 흐름제어를 이용하여 파일을 읽어서 다른 파일에 쓰는 예제

 

waterfall에서 처음 호출하는 함수에서는 fs.readFile을 이용해서 파일을 읽었다.

function(callback){                  fs.readFile(src,callback);               },

다음으로 fs.readFile에 대한 콜백 함수를 waterfall에서 넘겨주는 callback의 형태는 fs.readFile의 포맷이 fs.readFile( filename, function(err,data)) 형태이기 때문에 앞의 err 인자이외에 ‘data’ 인자만 필요하다.

그래서 다음에 오는 함수가 다음과 같이 ‘data 인자를 갖는 function(data,callback)이다.

function(data,callback){                  fs.writeFile(des,data,callback);               }

인자로 받은 data를 fs.writeFile에 넘겨서 파일을 쓰게 한다.

마지막 부분은 최종 콜백 함수로, 모든 함수가 실행이 끝나면 실행이 되고 또는 waterfall에 정의된 task 실행중에 에러가 나도 실행이 되는 부분이다. 첫번째 인자로 항상 err를 받는다. 에러가 없을 경우에는 이 값은 null 이된다.

function(err){                        if(err) console.log(err);                   }

본 예제에서는 파일 쓰기가 완료된 후에, 별도의 액션은 취하지 않아서 별다른 코드가 없지만, 에러가 발생했을때 처리하기 위해서 if(err)를 통해서 에러가 있으면 콘솔로 출력하도록 하였다.

 

series

series 흐름은, waterfall가 유사하게 정의된 task를 순차적으로 실행한다.

차이는 waterfall은 각 task에서 나온 결과를 다음 task의 입력으로 넘겼다면,

series는 각 task의 결과를 취합하여, 최종 callback에 배열 형태로 넘겨준다.

 

개념도를 보면 다음과 같다. series 흐름에 전달된 asyncfunctionA,B,C를 순차적으로 실행하고, 그 결과를 취합해서 맨 마지막 callback에 results라는 배열로 넘긴다.

waterfall과 마찬가지로 task수행중에 에러가 나면 실행을 중단하고, 최종 callback로 흐름을 옮기고, err에 에러에 대한 디테일한 내용을 기술해놓는다.

Figure 7 series 흐름 제어의 개념

이때 results 배열에는 task들의 결과값이 실행 순서대로 들어간다. 위의 그림에서 최종 callback으로 전달되는 results  배열에 asyncfunctionA에 대한 결과값 resultA, 두번째 인자는 asyncfunctionB의 결과값 B, 그리고 마지막 세번째 인자는 asyncfunctionC의 결과값 C가 들어간다.

 

series 흐름제어의 문법을 보자

series(tasks, [callback] )

·         tasks : 동시에 수행할 함수들을 배열로 정의

·         callback : 최종 callback으로, callback(err,results) 형태로 정의된다. 에러가 발생하면 err 변수에 에러에 대한 내용이 넘어오고, 정상적인 수행 완료인 경우에는 err는 null로 전달되고, 각 task의 실행 결과가 results 변수에 배열로 정의되서 리턴된다.

 

아래 예제는 위의 series 흐름을 async 프레임웍을 통해서 구현한 예제이다.

var async = require('async');   async.series([               function(callback){                  callback(null,'resultA');               },               function(callback){                  callback(null,'resultB');               },               function(callback){                  callback(null,'resultC');               }              ],              function(err,results){                        if(err) console.log(err);                        console.log(results)                              // handle resultC                   } );

Figure 8 sereis 흐름 제어를 사용한 예

실행하면 다음과 같은 결과가 나온다.

 

 

series 흐름은 서로 데이타에 대한 의존성은 없지만 순차적으로 실행이 되어야 하는 경우등에 활용이 될 수 있다.

예를 들어 사용자 정보가 MySQL과 MongoDB에 분산 저장되어 있고, MySQL에는 사용자ID와 암호화된 비밀번호를 저장하고, 기타 다른 정보를  MongoDB에 저장한다고 가정할때, 새로운 사용자 생성은 MySQL에 사용자ID등의 정보를 저장한 후에, MongoDB에 순차적으로 저장해야 한다면, 이 series 흐름이 유용하게 사용될 수 있다.

 

parallel

async 모듈에서 마지막으로 살펴볼 흐름제어는 parallel이다

이름에서도 볼 수 있듯이 동시에 여러개의 task를 실행하는 방법으로, 마치 멀티 쓰레드와 같은 효과를 낼 수 있어서, 실행 시간을 단축시킬 수 있다.

 

아래 개념 그림을 보자. 3개의 task를 병렬로 동시에 수행하는 개념이다.

asyncfunctionA,B,C를 동시에 수행하고 모든 작업이 끝나면 최종 callback을 수행한다.

수행결과는 최종 callback에 배열 형태로 전달된다.

Figure 9 parallel 흐름 제어의 개념

 

에러 처리는 parallel로 수행중이던 task중에 에러가 발생하면, 바로 최종 callback에 에러를 넘긴다. 단 이때 에러가 발생하지 않은 task들은 수행을 멈추지 않고 끝까지 수행되다.

 

parallel흐름 제어를 사용할때 주의해야 할점은 멀티 쓰레드처럼 작업을 수행해주는 것이지 실제 멀티 쓰레드가 아니다. IO작업등이 있는 task들의 경우 IO 요청을 보내놓고, 응답이 올때 까지 다른 task를 실행해서 병렬로 실행하는 것과 같은 효과를 주는 것이다. 만약에 task자체가 IO작업등이 없고 계속해서 CPU를 사용한다면, 그 작업이 끝난후에 다음 task로 넘어가기 때문에, 병렬 처리가 일어나지 않는다. (이런 경우에는 series를 쓰는게 나음)

 

parallel이 효과적으로 사용될 수 있는 곳은 IO쪽인데, 원격으로 여러개의 REST API를 동시 호출하거나, 또는 동시에 여러개의 쿼리를 조회하는 것들에 효과적으로 사용할 수 있다.

 

parallel 흐름 제어의 문법은 다음과 같다. series 흐름 제어 문법과 거의 동일하다고 보면 된다.

parallel(tasks,[callback])

Figure 10 parallel 흐름 제어 문법

·         tasks : 동시에 수행할 함수들을 배열로 정의

·         callback : 최종 callback으로, callback(err,results) 형태로 정의된다. 에러가 발생하면 err 변수에 에러에 대한 내용이 넘어오고, 정상적인 수행 완료인 경우에는 err는 null로 전달되고, 각 task의 실행 결과가 results 변수에 배열로 정의되서 리턴된다.

 

예제 코드를 살펴보자.

var async = require('async');   async.parallel([               function(callback){                  callback(null,'resultA');               },               function(callback){                  callback(null,'resultB');               },               function(callback){                  callback(null,'resultC');               }              ],              function(err,results){                        if(err) console.log(err);                        console.log(results)                              // handle resultC                   } );

Figure 11 parallel 흐름 제어 예제

 

parallel 흐름 제어의 문법은 series와 다르지 않다. 단지 내부 수행에 있어서 순차적으로 수행을 하는지 아니면 병렬로 동시에 수행을 하는지에 따른 차이만 있다. 위의 코드는 resultA, resultB, resultC를 내는 3개의 task를 동시에 수행하고, 수행이 끝나면, 최종 콜백 함수인 function(err,results)에서 results 배열에 결과를 출력하는 코드이다. 코드를 실행하면 다음과 같이 callback(null, 결과값)으로 넘긴 resultA,resultB,resultC 문자열이 출력 되는 것을 확인할 수 있다.

 

Figure 12 parallel 흐름 제어 예제 코드 실행 결과

 

실제 코드를 보면서 이해를 돕자.다음은 소스 코드 저장소인github에 bwcho75라는 사용자 정보를 조회하는 REST API와, bwcho75사용자의 follower를 조회하는 REST API 두개를 parallel을 이용해서 동시에 호출하여 결과를 화면에 출력하는 코드이다.

간단하게 REST 호출을 도와주는 모듈로 unirest를 사용하였다. http://unirest.io/

모듈을 사용하기 위해서 npm을 이용하여 코드를 작성하기 전에 unirest 모듈을 설치한다.

%npm install unirest

 

var async = require('async'); var unirest = require('unirest');   var start = new Date().getTime(); async.parallel([                 function(callback){                       unirest.get('https://api.github.com/users/bwcho75')                       .header('Accept', 'application/json')                       .header('User-Agent','mynodeapplication')                       .end(function(response){                            callback(null,response.body);                       })                                       },                 function(callback){                       unirest.get('https://api.github.com/users/bwcho75/followers')                       .header('Accept', 'application/json')                       .header('User-Agent','mynodeapplication')                       .end(function(response){                            callback(null,response.body);                       })                                   }                 ],                 function(err,results){                              console.log('Result 1 -------');                              console.log(results[0]);                              console.log('Result 2 -------');                              console.log(results[1]);                              console.log('elapsed time : '+(new Date().getTime() - start));       });

Figure 13 async의 parallel 흐름 제어를 이용하여 두개의 github REST API를 호출하는 예제

 

코드를 실행하면 다음과 같이 Result1, Result2에 대한 결과를 얻은것을 볼 수 있다.

Result 1 ------- { login: 'bwcho75',   id: 3168358,   avatar_url: 'https://avatars.githubusercontent.com/u/3168358?v=3',   gravatar_id: '',   url: 'https://api.github.com/users/bwcho75',   html_url: 'https://github.com/bwcho75',   : 중략   followers: 7,   following: 0,   created_at: '2013-01-02T10:33:11Z',   updated_at: '2016-02-02T02:38:53Z' } Result 2 ------- [ { login: 'yshu0307',     id: 1740343,   : 중략     type: 'User',     site_admin: false },   { login: 'kmoonki',     id: 1725366,   : 중략     type: 'User',     site_admin: false },   { login: 'z-n',     id: 5715797,   : 중략     type: 'User',     site_admin: false },   : 중략 ] elapsed time : 1915

Figure 14 parallel 흐름 제어를 이용하여 두개의 github REST API를 실행한 결과

병렬로 API가 호출되었는지를 확인 하기 위해서, 맨 마지막 부분에 API 호출에 소요된 시간을 출력하였는데, 예제 코드에서 async.parallel을 async.series로 바꿔서 호출해보면 순차 호출로 바뀌게 되는데, 수행시간이 본인이 테스트한 경우 200ms정도 더 나왔다. 즉 병렬 호출을 통해서 200ms 정도의 수행 시간을 절약한것이다.

 

지금까지 async 모듈을 이용하여 콜백헬을 해결하고, 제어 흐름을 컨트롤할 수 있는 방법에 대해서 알아보았다. async 모듈에서 위의 3가지 흐름제어가 많이 사용되기는 하지만 이외에도 많은 흐름 제어 방식이 있기 때문에, https://github.com/caolan/async 를 참고하기 바란다.