core.async

core.async is the official library for asynchronous programming with channels, in the style of Go's goroutines (Communicating Sequential Processes). Processes put values into channels and take values out; channels coordinate them without shared mutable state.

Add it with {:deps {org.clojure/core.async {:mvn/version "1.9.865"}}} (the examples used 1.9.865).

Channels and go blocks

(require '[clojure.core.async :as a :refer [chan go >! <! >!! <!! close!]])

(def c (chan))

(go (>! c "hello from a go block"))
(println (<!! c))

(def results (chan 10))                ; a buffered channel
(doseq [n (range 5)]
  (go (>! results (* n n))))
(println (sort (repeatedly 5 #(<!! results))))

hello from a go block
(0 1 4 9 16)

• (chan) makes an unbuffered channel: a put waits until someone takes. (chan 10) buffers up to 10 values.
• go runs its body asynchronously on a small thread pool. Inside, >! (put) and <! (take) park the go block instead of blocking a thread, so you can have thousands of them.
• >!! and <!! are the blocking versions, for use outside go blocks (like the main thread here).
• The five go blocks finish in any order, so the example sorts the results.

A producer–consumer pipeline

(require '[clojure.core.async :as a :refer [chan go go-loop >! <! <!! close!]])

(def jobs (chan 5))
(def done (chan))

(go
  (doseq [j (range 1 6)]
    (>! jobs j))
  (close! jobs))

(go-loop [total 0]
  (if-let [j (<! jobs)]              ; nil means the channel is closed
    (recur (+ total (* j 10)))
    (>! done total)))

(println "total:" (<!! done))

total: 150

Taking from a closed channel returns nil once it's empty, which is how consumers know to stop. go-loop is (go (loop ...)).

Timeouts and alts!

(require '[clojure.core.async :as a :refer [chan go <! >! timeout alts! <!!]])

(defn fetch [ms value]
  (let [c (chan)]
    (go (<! (timeout ms)) (>! c value))
    c))

(defn first-or-timeout [c ms]
  (go
    (let [[v port] (alts! [c (timeout ms)])]
      (if v v :timed-out))))

(println (<!! (first-or-timeout (fetch 50 :fast-reply) 500)))
(println (<!! (first-or-timeout (fetch 1000 :slow-reply) 100)))

(let [[v _] (<!! (go (alts! [(fetch 300 :mirror-a) (fetch 30 :mirror-b)])))]
  (println "fastest:" v))

:fast-reply
:timed-out
fastest: :mirror-b

• (timeout ms) is a channel that closes after ms milliseconds.
• alts! waits on several channels and returns [value channel] for whichever is ready first — the basis for timeouts, racing requests, and cancellation.

Channels with transducers and pipelines

(require '[clojure.core.async :as a])

(def evens-squared (a/chan 10 (comp (filter even?) (map #(* % %)))))
(a/onto-chan! evens-squared (range 10))
(println (a/<!! (a/into [] evens-squared)))

(def in (a/to-chan! (range 1 9)))
(def out (a/chan))
(a/pipeline 4 out (map #(* 100 %)) in)     ; 4 workers, order preserved
(println (a/<!! (a/into [] out)))

[0 4 16 36 64]
[100 200 300 400 500 600 700 800]

A channel can carry a transducer (from Level 2) that transforms values as they pass through. pipeline processes values in parallel while keeping their order.

Guidelines

• Never do blocking I/O inside a go block — it ties up one of the few go threads. Use a/thread (or a/io-thread in recent versions) for blocking work and return its channel.
• Always decide who closes a channel (usually the producer).
• For simple "run this in the background and get the result later", a future is enough; reach for core.async when several processes need to communicate.
• core.async also works in ClojureScript, where it tames callbacks.