Clojure Guide — core.async: Asynchronous Programming

In this tutorial, you will learn about Clojure Guide. We cover key concepts, practical examples, and best practices to help you master this topic.

core.async is a Clojure library that implements Communicating Sequential Processes (CSP) using channels and go blocks, enabling asynchronous programming with the same mental model as goroutines in Go but fully within Clojure's functional paradigm.

What You'll Learn

• Channels for communicating between processes
• Go blocks for lightweight asynchronous execution
• Parking vs blocking operations
• Buffered and unbuffered channels
• Coordination patterns with alt! and alts!

Why It Matters

Asynchronous programming is essential for modern applications. core.async provides a clean, composable model for async workflows without callbacks or futures. Durga Antivirus Pro uses core.async for coordinating concurrent file scans.

Real-World Use

Netflix uses Clojure core.async for Stream Processing. CircleCI uses it for build pipeline coordination. core.async powers real-time data processing and event-driven systems.

flowchart LR
    A["core.async"] --> B["Channels"]
    B --> C["Go Blocks"]
    C --> D["Thread Channels"]
    D --> E["Coordination"]
    A:::current --> B
    style A fill:#2563eb,stroke:#2563eb,color:#fff
    style B fill:#dbeafe,stroke:#2563eb,color:#1e40af
    style C fill:#dbeafe,stroke:#2563eb,color:#1e40af
    style D fill:#dbeafe,stroke:#2563eb,color:#1e40af
    style E fill:#f1f5f9,stroke:#94a3b8,color:#64748b

Creating Channels

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

;; Unbuffered channel
(def ch (chan))

;; Buffered channel
(def buffered-ch (chan 10))

;; Channel with transducer
(def xf-ch (chan 10 (filter number?)))

Go Blocks

Go blocks create lightweight asynchronous processes that run on a thread pool:

;; A simple go block
(go
  (let [result (<! ch)]
    (println "Received:" result)))

;; Multiple go blocks
(go
  (>! ch "hello"))

;; The entire block runs asynchronously

Parking Operations

<! and >! are parking operations that suspend the go block without blocking a thread:

(def ch (chan))

(go
  (println "Sending...")
  (>! ch "data")        ; Park until someone receives
  (println "Sent!"))

(go
  (println "Receiving...")
  (let [data (<! ch)]   ; Park until someone sends
    (println "Got:" data)))

Blocking Operations

<!! and >!! are blocking operations for non-go-block contexts:

;; From regular thread (not go block)
(def ch (chan))

(future
  (Thread/sleep 1000)
  (>!! ch "from future"))

(println "Waiting...")
(println "Got:" (<!! ch))
;; => Waiting...
;; => Got: from future

Buffered Channels

;; Buffered channel won't block sender
(def ch (chan 3))

(go
  (>! ch :a)
  (>! ch :b)
  (>! ch :c)
  (println "All sent without blocking"))

;; Buffer overflow
(doseq [i (range 10)]
  (>!! (chan 5) i))  ; blocks after 5 items

Channel Coordination

;; alt! for selecting from multiple channels
(def ch1 (chan))
(def ch2 (chan))

(go
  (alt!
    ch1 ([v] (println "Got from ch1:" v))
    ch2 ([v] (println "Got from ch2:" v))
    (timeout 1000) ([_] (println "Timed out!"))))

;; alts!! for blocking version
(let [[v ch] (alts!! [ch1 ch2 (timeout 1000)])]
  (println "Got" v "from" ch))

Pipelines

;; Pipeline pattern
(defn pipeline [n]
  (let [in (chan)
        out (chan)]
    ;; Multiple workers
    (dotimes [_ n]
      (go (let [v (<! in)]
            (when v
              (>! out (process v))
              (recur)))))
    [in out]))

(defn process [x] (* x 2))

Error Handling

(go
  (try
    (let [data (<! ch)]
      (process data))
    (catch Exception e
      (println "Error:" (.getMessage e)))))

Common Mistakes

1. Blocking in go blocks

Never use >!! or <!! inside a go block. Use parking operations >! and <! instead.

2. Not closing channels

Always close channels when done to prevent resource leaks and signal completion to consumers.

3. Channel starvation

If a producer is faster than a consumer, an unbuffered channel causes contention. Use buffered channels or add more consumers.

4. Assuming go blocks are threads

Go blocks are not threads. They are state machines that park at <! and >!. Long-running CPU work in a go block blocks the thread pool.

5. Deadlocks with coordinated operations

Multiple go blocks waiting on each other can Deadlock. Use alt! with timeouts to prevent hangs.

Practice Questions

1. What is the difference between <! and <!!? <! parks the current go block (non-blocking thread). <!! blocks the actual thread. Use <! inside go blocks and <!! outside.

2. What happens when a buffered channel is full? The sender blocks (or parks) until a receiver takes an item, freeing buffer space.

3. How does alt! prevent deadlocks? alt! can include a timeout clause, ensuring the operation completes even if no channel is ready.

Challenge: Implement a fan-out pattern where one producer sends to multiple workers and collects results.

FAQ

{{< faq question="Is core.async similar to Go's goroutines?" >}} Yes, the CSP model is based on the same concepts. Go blocks are similar to goroutines, and channels work identically. The main difference is that Clojure's channels are first-class values. {{< /faq >}}

{{< faq question="How many go blocks can I create?" >>> Thousands. Go blocks are very lightweight (similar to goroutines). Each go block uses minimal memory for its state machine. {{< /faq >}}

{{< faq question="Can I use core.async with transducers?" >}} Yes. You can attach a transducer to a channel when creating it, transforming each element as it passes through. {{< /faq >}}

{{< faq question="What is a parking operation?" >}} A parking operation suspends the go block without blocking the underlying thread. The thread is freed to run other go blocks, enabling efficient concurrency. {{< /faq >}}

{{< faq question="How do I handle errors in go blocks?" >}} Use try/catch inside the go block. Errors outside go blocks can be caught at the channel reader with Exception Handling. {{< /faq >}}

Mini Project

Build a simple task queue with worker pool:

(defn task-queue [workers]
  (let [tasks (chan 100)
        results (chan 100)]
    (dotimes [i workers]
      (go (while true
            (let [task (<! tasks)]
              (>! results (str "Worker " i " processed: " task))))))
    [tasks results]))

(let [[tasks results] (task-queue 3)]
  (doseq [i (range 10)]
    (>!! tasks (str "task-" i)))
  (close! tasks)
  (dotimes [_ 10]
    (println (<!! results))))

What's Next

Now that you understand core.async, explore Datomic for immutable database design.