Clojure + core.async による非同期＆並列プロセスの世界
core.asyncによる非同期プログラミング
core.async はClojure用の、事実上標準の非同期プログラミングのライブラリです。
core.asyncの一番わかりやすい説明は、「Go-langのchannelのClojure版」という言い方でしょう。goマクロによってgo-blockを作り、そのブロック内が非同期に動きます。このブロックが常駐すれば、軽量プロセスというやつになります。プロセス同士のやりとりをする口として、チャネル(channel)があります。core.asyncを使ったプログラムでは、チャネルへの入出力を介して非同期軽量プロセスにデータを処理させることで、全体のシステムを作り上げます。
goマクロはステートマシンを作り、チャネルへの入力があるたびにマシンが１回転します。この一回転時に、チャネルを待ち受けていたgoブロックにスレッドが割り当てられ、次のチャネル入出力までCPUを使って処理が動き、チャネルの入出力でまた別のgoブロックに処理が映り、という形で、限られたCPU上で、スレッドを山ほど起動することもなく、効率よく動作するのが売りの一つです。
このような仕組み（OSの協調型マルチプロセスと同じような原理）なので、goブロックは実際にはプロセスでもスレッドではなく、ステートマシンによって管理されたプログラム単位に過ぎません。よって、core.asyncはスレッドが一つであってもちゃんと動きます。core.async開発当初から、ClojureScript（JavaScriptをホスト言語としたClojure実装）でも動くことを想定して作っていたということですので、ならではの実装でしょう。
シンプルな仕組み
core.asyncの使い方については公式ドキュメントとかのほうが詳しいので詳細はそちらを見てもらうとして、簡単に概要だけを書くと、実行単位をgoで囲み、そのなかでchannelを読んだり書いたりすると、goで書かれた実行単位が次々と切り替わって実行されます。
以下は、CloureのREPLに打ち込めばそのまま動作する、core.asyncを使ったプログラムコードです。
(import '[java.util Date]) (require '[clojure.core.async :refer [chan go-loop >! <! timeout] :as async]) (def ch (chan)) ; チャネルを作る ;; 書き込み非同期ブロック (go-loop [] (when (>! ch (Date.)) ; チャネルに書く (<! (timeout 2000)) ; 2秒待つ (recur))) ;; 読み込み非同期ブロック (go-loop [] (when-let [date (<! ch)] ; チャネルを読む (println "now:" date) (recur)))
go-loop は
(go (loop [] ;; 処理 ))
の省略形で、多用するので用意されています。
単純なgoブロックは、非同期処理が終わるともう２度と実行されません。しかし、loopすれば「ずっと動き続ける非同期ブロック」を作れます。これが「軽量プロセス」に近いものです。軽量プロセス同士がチャネルを使ってデータをやりとりする、というプログラムを作るには、毎回goとloopを書かなければいけなくて、少々面倒なので、二つをまとめたgo-loopマクロが用意されています。
このプログラムは、片方のgoブロックが現在日時をチャネルに書き込んで2秒待つ、もう一つのgoブロックは同じチャネルを読み込み、読めたらそれを画面に出力します。いずれのgoループもチャネルが閉じられるまでloopし続けます。現在時刻を生成するプロセスと、受け取った時刻を出力するプロセスの、２つの軽量プロセスが動いていると考えればいいでしょう。
このように、core.asyncでは、プログラムを処理単位ごとにgoブロックで囲って非同期処理にし、そのgoブロック間でのデータのやりとりにはチャネルを使います。
goブロックは、チャネルからデータを読み書きしようとして、もしチャネルにまだデータがなかったり、チャネルにまだ書けない状態だったら、park（待機）状態になりスレッドを解放します。そして準備ができればまたスレッドに割り当てられて動き出します。
非常に少ないスレッド数で、たくさんの非同期ブロックを実行できるわけです。
複雑なスレッド制御を書かなくとも、シンプルにチャネルを読み書きするところをgoで囲むことで、簡単に効率よい並列プログラムを書けるのがcore.asyncの強みです。プログラムを書く側は単純にやりたいことを上から下へ書いていくだけでいいのです。callback hellと呼ばれるような、非同期コールバック関数が何段にも重なるようなことはありません。goブロックを上からどんどん書いていけばいいのです。
と、ここまでは、core.asyncにもともと備わっていた基本機能です。ここに、Clojure 1.7の言語拡張により、新しい要素が追加されました。この機能により、core.asyncは一段と便利になりました。
Transducerの登場
core.asyncのために、Clojureの言語レベルでの拡張までも行われました。それが、Clojure 1.7でのTransducersの導入です。
Transducerというのはおおざっぱに言うと、map処理やfilter処理から、対象となるオブジェクト（コレクション）を省いて、変換処理だけを抜き出して抽象化したものです。(map change-fn coll)という処理なら、重要なものは「ひとつひとつの要素にchange-fnを適用する」という変換処理であって、collは引数に過ぎない、だったらその変換処理部分だけを抜き出して別のオブジェクトとして扱えるようにしよう、というわけです。
実際、transducerの作り方は、上の説明通りのものです。
(map change-fn coll) ; いつものmap処理。collの各要素にchange-fnを適用する遅延リストを作り出す
(let [xf (map change-fn)] ; 同様の処理を行うtransducerを生成する。引数は後から渡すことができる。 (sequence xf coll)) ; collの各要素にxfというtransducerを適用する遅延リストを作る
いつものmapやfilter関数呼び出し時に、対象となるcollectionを渡さなければ、変換処理だけを取り出したtransducerになるのです。
さらに、transducerは合成することもできます。transducerは特定のルールに則って実装された関数に過ぎないので、Clojure標準の関数合成関数 comp で簡単に合成できます。
(comp 第1のtransducer 第2のtransducer 第3のtransducer)
ただの関数をcompした場合、一番後ろの関数から順に実行されます。しかしtransducerの場合、その構造上、実際の実行は頭から行われます。上の例では、第1のtransducerから順に、3まで実行するような、合成transducerができます。
(require '[clojure.string :as string]) ;; string/upper-caseで大文字にmapした後、 ;; T以外のものにfilterするtransducerを作る (def xf (comp (map string/upper-case) (filter #(not= % "T")))) ;; sequenceは引数にtransducerを適用したリストを作る ;; 大文字に変換された後、Tでない文字だけにfilterされます。 (apply str (sequence xf "TEST")) ; => "ES"
なぜこのようなアイデアが用意されたかというと、（経緯はいろいろあるんでしょうが、私の認識では）core.asyncの開発途上で必要性が認識されたからです。もともとcore.asyncには、チャネルに入出力するデータに対してmapやfilterを実行するための、専用の関数がたくさん用意されていました。 clojure.core.async/map> とか。でも、標準のmap関数との違いは、処理対象がコレクションかチャネルか、という違いだけで、実際にやりたいこと（mapしたい、filterしたい）は同じなわけです。だったら、その同じ部分だけを抜き出そうというのは自然な発想です。
というわけで、Clojure 1.7にはTransducerが導入され、mapやfilterといったもともとあった多くの関数が拡張されて、いままでの処理のほかに、transducerを作り出す機能が追加されました。
同時に、core.asyncにあった、専用のmapやfilterといった関数群は、deprecated扱いとなりました。代わりに、チャネルに対してtransducerを設定することができるようになりました。これにより、チャネル入出力＝関数の実行、という世界ができあがりました。
現在のcore.asyncは、単なるClojure版goルーチン実装の域を超えて、transducerの導入により、関数実行エンジンとしての機能を備えました。
;; 入力した文字列を大文字化するtransducerを設定したチャネルを生成する (require '[clojure.string :as string] '[clojure.core.async :refer [chan go >! <!]]) (let [ch (chan 1 (map string/upper-case))] ; 大文字化するtransducerをセットしたチャネルを作る ; 小文字を書き込んでみる (go (>! ch "test")) ; 読み込んで出力すると、大文字になっている！ (go (println "result:" (<! ch)))) ;;=> result: TEST
core.asyncには、チャネルとチャネルを結合する pipe という関数があるので、これを使って変換処理をつなげることもできます。
(require '[clojure.core.async :refer [chan go-loop >! <! pipe onto-chan]]) (let [only-odd-ch (chan 1 (filter odd?)) ; 奇数だけを通すチャネル double-ch (chan 1 (map #(* % 2))) ; ２倍にするチャネル ch (chan) piped (-> ch (pipe only-odd-ch) ; ch を only-odd-ch に連結する (pipe double-ch))] ; 前行の連結結果をさらに double-ch に連結する ;; onto-chan関数は内部でgoブロックを使ってコレクションの中身を非同期にチャネルに書き込む (onto-chan ch [1 2 3 4 5 6]) ;; 連結した末尾にあるpipedから、非同期にデータを一つずつ読む (go-loop [] (when-let [data (<! piped)] (println "data:" data) (recur)))) ;; 奇数の1, 3, 5 が2倍になって順番に出力される ; => data: 2 ; => data: 6 ; => data: 10
pipeによってチャネルを結合してtransducerを順次実行していくことができるわけです。
しかしこれだけでは、関数を順次実行しているだけで、直接関数を呼ぶのに比べて、めんどくさくなってるだけではないでしょうか。
実はこれだけでも、関数を直接呼び出すのとは違う利点もあるのですが、そこは一旦置きましょう。core.asyncは並列実行ライブラリです。transducerの実行を並列化しましょう！
pipelineによる関数の並列化
transducerの登場で、データの変換操作を簡単に抽象化することができるようになりました。チャネルの入出力と関数を紐付けることが可能になったわけです。
ここでpipelineが登場です。
pipelineというのは、その名の通り（チャネルとチャネルの）パイプラインを構築する関数です。パイプラインの構築には、入力と出力のチャネルに加えて、実行したい関数、それに並列数を指定できます。
pipeline関数とは、チャネルからチャネルにデータを転送するときに関数を適用するという処理の、「関数を適用する」部分を並列化しよう、というものです。
(pipeline 4 out-ch my-great-transducer in-ch)
この１行で、in-chからout-chへのパイプラインが構築されて、in-chにデータを入れると、out-chから出力されます。その途中に、my-great-transducerによりデータが処理されるのですが、このtransducerの実行は、最大で４並列で処理されます（ちなみに、out-chへはin-chにデータが入ってきた順序で結果が出力されることが保証されているので、順番は壊れません）。チャネルに立て続けに４つデータが入ってくれば、それらは並列で処理されるということです。
チャネル、transducer、pipelineの3つで、core.asyncの役者がそろいました。core.asyncはClojureでgoブロックを実現するライブラリでありつつ、その上に、「transducerという変換関数を並列実行するpipelineを構築して、そのpipeline同士をつなげることで並列プログラムを構築する」という世界ができてるわけです。
この仕組みは、関数によるプログラムモデルにもよい意味での影響を与えます。
たとえば、たくさんのリクエストを受け付けるサーバプログラムで、データ処理プロセスが一つ存在して、それに処理を依頼するようなプログラムモデルを考えてみましょう。サーバプログラムは、たくさんのリクエストを並列に受け付けますから、このエンジンももちろん並列で動いてほしいです。ならば、データ処理を行うpipelineを構築して、その入力チャネルにデータを入れればいいのです！ pipelineが、並列処理を実行する軽量プロセスとなるのです。
pipelineは入力がない限り待機してCPUを消費しませんが、入力チャネルに書き込めばいつでも動きます。go-loopで処理を繰り返すのと同じことを、pipelineは行っています。ただ、go-loopは処理を非同期化してくれはしても、並列化はしてくれません。pipelineは、transducerを並列に実行してくれます。
効率的なデータ処理エンジンを構築する
このプログラムはサーバプログラムなので、たくさんのリクエストを外部から受け付けます。秒間100とかもっととか、とにかく並列にたくさんの要求を受け付けます。
一方でサーバのCPU数には限りがあるわけで、すべてのリクエストごとにデータ変換関数を無制限には実行したくありません。一度に実行する変換処理は、例えばCPUコア数までとかに絞りたいわけです。そこで、プログラムの中心に「データ変換エンジン」と呼ぶ、入出力を受け付けるプロセスを用意します。リクエストを受け付けたら、このエンジンにデータを投入すると、並列にデータ変換を実行し、最後にレスポンスを返すものとしましょう。
リクエストを受け付けるサーバ自体の実装は今回のテーマではないので無視することにして、とりあえず read-ch を読み込むとリクエストが読み込めて、 write-ch に書き込むとレスポンスが返る、ということにしておきましょう。
用意するものは
- リクエストをパイプラインに流し込む関数（始端処理）
- データを変換する関数
- レスポンスを出力する関数（終端処理）
これだけです。パイプラインには必ず始まりと終わりがあるので、はじまりには何かしらの始端処理が、終わりには何かしらの終端関数があるはずで、今回の場合の始端処理はリクエストをエンジンに渡す（パイプラインに流す）ような何かで、終端処理は、「レスポンスを返す」ということになります。終端処理は、たいていの場合は、パイプラインの最後の出力を読み取って、それを（ファイルとかネットワークとか）どこかに書き出す、という処理を行うような、goブロックです。
エンジンを作る
データを変換する関数の内容自体は今回は重要ではないので、とりあえず great-convert-xf というすごいtransducer関数があることにします。多分、すごい関数をtransducer化してさらにcompを使って合成したようなものです。これを使って並列パイプラインを作ります。
(require '[clojure.core.async :refer [chan pipeline]]) (def in-ch (chan)) (def out-ch (chan)) (def engine (pipeline 8 out-ch great-convert-xf in-ch))
なんと今回はこれでエンジンは完成ですね。engineと名付けたこのpipelineは、in-chにデータが書き込まれると、great-convert関数を最大8並列で実行してくれます！
もちろん今回は一番大変な great-convert-xf transducerの実装を省略しているから、これで済んでいるわけですが、関数さえ存在すれば、それを並列エンジン化するのは簡単だということがわかると思います。
始端処理
始端処理が行うべきは、ネットワークからのリクエストが入ってくるread-chを読み取って、エンジンの入力であるin-chに流し込むだけです。 ネットワークから来るデータが、そのままエンジンに流し込めるデータ構造の場合は、とても簡単です。
(pipe read-ch in-ch false)
read-chとin-chを直接つなげちゃえばいいのです！ 最後の引数falseは、read-chがcloseされた場合に接続先チャネル(in-ch)をcloseしない、という意味です。これを忘れると、一個のリクエストを処理したところでエンジンの入力チャネルが閉じられてしまうので注意です。
ですが実際には、ネットワークから来るデータは、バイナリだったり、JSON文字列だったりするので、エンジンが処理できるデータ形式に変えてから流し込むことになります。go-loop関数で実現できます。
(require '[clojure.core.async :refer [go-loop <! >!]]) (go-loop [] (when-let [req (<! read-ch)] (when (>! in-ch {:data (convert-request req) :write-ch write-ch}) (recur))))
このgo-loopは、read-chが閉じられる（読み取り結果がnilになる）か、エンジンの入力チャネルが閉じられる（書き込みでfalseが返る）かするまで、無限にループします。ポイントは、これはgoブロックなので、無駄にループしてCPUを消費しないことです。チャネルから読み取れるデータがあれば、core.asyncによってgoブロックに処理スレッドに割り当てられます。そして次のチャネル入出力時に再び休止状態に戻り、ほかのgoブロックに処理が回ります。CPUを効率的に利用できるのです。
エンジンに流し込むデータは、加工済みデータと、レスポンスを出力するためのチャネルの、二つの要素の入ったマップです。
終端処理を作る
終端処理は始端処理の逆ですので、似たような処理となります。エンジンからは、エンジンに流し込んだのと同じ形式のマップとしてデータが出力されるものとします。
(go-loop [] (when-let [{:keys [data write-ch]} (<! out-ch)] (>! write-ch data) (recur)))
この終端処理は、エンジンの出力チャネル(out-ch)が閉じられるまで、無限にループします。始端処理と同じく、ループによってCPUを無駄に消費することはありません。
これで、始端処理→エンジン→終端処理というサイクルができあがりました。エンジンはcore.asyncのpipelineを使うことで並列に実行されます。エンジンが処理する関数がひとつだけの、とてもシンプルな例ですが、core.asyncで並列処理サーバプログラムを作るときの基礎がちゃんと入っています。
pipelineを拡張する
ポイントは、一度この構造ができてしまえば、エンジンの部分は容易に変更できるということです。プログラムの大枠として、始端→エンジン→終端、という構造さえできていればよくて、エンジンの部分をもっと拡張しても、この構造さえ変わらなければ問題ないのです。
そりゃそうだろ、エンジン部分のプログラムが難しいんであって、そこを書いてないのだから、というのはその通りです。しかしポイントは、プログラムの拡張を、core.asyncのパイプラインの拡張という形で行えるというところです。
関数は密結合である！
Clojureの作者であるRich Hickeyが、core.asyncについて説明した プレゼンテーション があります。この動画をみると、core.asyncはただgoルーチンをclojureで実現したものというのとは別の観点もあることがわかります。Richは、このプレゼンテーションの中で、（オブジェクト指向の利点と欠点と対比しつつ）関数プログラムの利点と欠点を簡単に話しています。ここでRichが関数の利点としているのは、関数はロジックを抽象化できるということ（一方、オブジェクトはロジックというよりはそれ自体がマシンである、としてます）、欠点は、関数はどうしても密結合になりがちだ、ということです。
関数呼び出しの連鎖を分解して、途中に分岐（if文とか）を挟み込むのは、結構骨の折れる作業です。ロジックが変わるから事実上検証もやり直しです。
そこで、Richは、関数（データの入力と出力がある）を、キュー（データを入力して出力するという機能しかない単純な構造）と組み合わせることで、関数同士の密結合をほどくことができる、と言っています。
それゆえ、Richは、仮にcore.asyncの非同期機能を一切使わずに、キュー（チャネル）を介して関数と関数を結びつけるだけでも、利点はあると言っています。
データ処理エンジンは、複雑な分岐を含んだ巨大な関数です。これを、pipelineを使って、容易に再接続可能な関数の集まりとして作れるのです。何しろ、pipelineというのは、チャネルという名のキューを介して関数同士を接続するためのものだからです。
パイプラインに分岐を組み込む
パイプラインの入出力はチャネルなので、たくさんのパイプラインを用意して、あるパイプラインの出力チャネルを別のパイプラインの入力チャネルとして指定すれば（あるいは、pipe関数で互いの入力と出力チャネル同士をつなげれば）、パイプラインをつなぎ合わせた巨大なパイプラインを作ることができます。
さらに、core.asyncには、チャネルに入ってくるデータを、別の複数のチャネルに、条件によって振り分ける関数が用意されています。pubとsubです。
ある出力用チャネルにpub関数を適用すると、publicationという特殊なオブジェクトを作れます。publicationは、チャネルに入ってくるデータの種別を識別することができます。さらに、別のチャネルをsub関数を使ってpublicationに登録することができます。この登録時に「このチャネルには、データの種別がAのデータだけを流してください」という指定ができるのです。
たとえば、パイプラインに入ってくるデータが「新規データ」「更新データ」「削除データ」に分かれているケースを考えてみましょう。この三種類それぞれ、行うべき作業は微妙に異なります。こういうとき、pub/subを使って、チャネルを三つに分岐させることができます。
(let [publication (pub ch :type)] (sub publication :new new-data-ch) (sub publication :update update-data-ch) (sub publication :delete delete-data-ch))
イメージとしてはこんな感じです（雑な手書き図ですみません）
ここで、「もともとは新規と削除しかないと聞いてた」→「急に、『実は更新データもありました…』とか言われてしまった」というケースを考えてみます。
core.asyncでは、もともとの「新規」用パイプラインと、「削除」用パイプラインに影響を与えずに、「更新」用パイプラインを追加することができます。単に、「更新」用パイプラインを作って、その入力チャネルを、分岐publicationに追加すればいいのです。
core.asyncは名前からも非同期処理で注目されますが、「密結合した関数呼び出しを疎結合にする」という目的も入っているのです。既存の関数（pipelineにtransducerとして組み込まれている）は変更されないので、この部分の再テストは不要です。新しい関数を用意し、テストし、transducer化し、pipelineを構築する。あとはチャネルとチャネルをどう接続するか、というだけの問題です。チャネル同士の接続を工夫することで、関数の分岐や実行順序を制御できるわけです。しかも、各関数は効率的に並列実行されるのです。
バイパスを作る
異常が発生したケースなど、正常なプログラムの流れを無視して、別の処理にジャンプしたいというケースは結構あります。実際、プログラミング言語の例外とキャッチというのは、正常なプログラムを流れを無視して例外処理へとジャンプする処理な訳で、関数をつなげてプログラムの流れを作るパイプラインであっても、やはり同様のことが必要になるケースはあり得ます。
core.asyncでは、pipelineで実行される関数は並列に実行される（別スレッドで動く）ので、単純に例外処理を行うことができません。その代わりに、exception-handlerと呼ばれる例外処理用の関数を使うことができます。exception-handlerは、引数として例外一つを受け取る関数であれば、なんでも構いません。
pipelineに設定したtransducer内でエラーが発生すると、（あれば）exception-handler関数が呼ばれます。exception-handlerは何をしてもよいですが、exception-handler関数の結果が、transducerの実行結果になります。
;; exception-handlerを設定したpipelineを作る例 (let [ex-handler (fn [ex data] (handle-error ex) nil)] (pipeline 8 out-ch (map my-greate-fn) in-ch false ex-handler))
ここでは使いませんが、実は、チャネルを作成する chan 関数にも、exception-handlerを渡すことができます。チャネルにtransducerをセットした時に、transducer内で発生したエラーを処理するために使います。
通常、エラーとなったデータはパイプラインの先に進めたくありません。そういうときは、exception-handler関数の戻り値をnilにします。core.asyncのチャネルにはnilを値として流すことができない仕様ですので、exception-handlerがnilを返すと、そのデータはパイプラインの先には流れません。 代わりに、exception-handlerに別の出力用チャネルを渡しておき、そこにデータを流すのです。このチャネルの先には、例外を処理するためのpipelineを設定しておきます。
;; バイパス用channelとしてbypass-chがすでにあるものとする ;; このex-handler関数は、受け取った例外をbypath-chに転送する。 (defn ex-handler [ex] (go (>! bypass-ch ex)) nil)
このようなexception-handlerをすべてのpipelineに設定しておけば、例外はすべて、bypass-chを経由して、例外処理用pipelineへと流れていくことになります。エラーのためのバイパスを作ったわけです。
core.asyncにはpub/subによる分岐とは別に、mergeによる連結機能もありますから、バイパスに流しておいて処理した後、最終的には同じ終端処理に接続する、といったことも可能です。
;; 終端処理につながる last-ch というチャネルがあるとする ;; すべてのチャネルを連結したall-data-chを作り、それをlast-chに ;; 連結する (let [all-data-ch (merge [new-ch update-ch delete-ch bypass-ch])] (pipe all-data-ch last-ch))
例外を処理するpipelineまで含めると、このシステムの全体像はこんな感じになります。
この柔軟性が、core.asyncの強みです。関数をtransducerとしてpipelineに組み込むことで並列化し、さらに、チャネルというキューを介して、pipelineを柔軟に結合して、システム全体を作るのです。Clojure自身の「データは基本的に不変である」という性質が、非同期処理の安全性を高めてくれています。入力と出力しかない「関数」と、同じく入力と出力しかない「キュー」を使うことでデータの流れを作り、さらにここに、このtransducerという変換処理を組み合わせると、pipelineという並列化関数を作り上げられるのです。これらを自由に組み替えられるわけです！
まとめ
ちょっとした例ですが、パイプラインをつなげて、分岐を伴うプログラムを作り上げるイメージができたでしょうか。
- goとchannelの二つで、非同期ブロック間のデータのやりとりを実現する
- transducerにより変換処理を抽象化し、合成可能にする
- channelとtransducerを組み合わせて、transducerを並列実行するpipelineを作る
- pipeline同士を、channelを介して分岐したり結合したりしてつなぎ合わせる
- データがパイプラインを終端に向かって流れていくとき、すべての処理は自動的に並列に実行される！
go（非同期ブロック）, channel（キュー）, transducer（変換処理）, pipeline（処理の並列実行）という４つの概念を組み合わせることで、並列プログラムができてしまいました！
それぞれが一つの仕事だけを行うような物を、組み合わせて複雑な処理を作り上げる、しかもそれらを混ぜ込まない（コンプレクトさせない）というのは、Clojureの目指す「Simple Made Easy」の世界観とも合っていますね。core.asyncは、Clojureらしい形で、並列プログラムの作り方を変える仕組みを実現した、Clojureのキラーライブラリの一つです。
core.asyncには、他にも、チャネルに入ってきたデータを複数のチャネルに（同じデータを）転送する multiple や、複数のチャネルをマージしつつ、チャネル単位で流れを一時的に停止できる mix、複数のチャネルから最初にデータが入ってきたものを選択できる alt! など、チャネルをつなぎ合わせるための関数が用意されています。これらを使えば、柔軟にチャネルの接続を動的に切り替えるなどの処理もできます。チャネルの組み合わせと並列化こそがcore.asyncの強さであり、ただのgoブロックではないということが分かるでしょうか。
宣伝
clj-ebisu で、core.asyncを使った時の「あるある」なトラブルと回避方法について、ちょっとしたプレゼンテーションをすることになりました。きてね。
