Главная мысль урока: в core.async части программы не делят общее состояние, а передают друг другу значения через каналы. Канал это очередь с правилами ожидания, go-блок это лёгкий процесс, который ждёт на канале, не занимая поток. Отсюда и главное правило: внутри go не блокируют.
В «Хрониках Пограничья» бой порождает поток событий: удар, промах, лечение, смерть монстра. Их читают сразу три части игры. Журнал пишет строку в events.log. Система достижений ищет «десять ударов подряд». Интерфейс рисует полоску здоровья. Можно завести атом со списком событий и заставить всех троих его опрашивать. А можно сделать, как в Go: бой кладёт события в трубу, читатели забирают их с другого конца, и никто не знает о существовании остальных.
бой ──▶ [ канал событий ] ──▶ журнал
└──▶ достижения
└──▶ интерфейсЭта труба называется каналом, а сама идея, что процессы общаются сообщениями, а не общей памятью, называется CSP. Библиотека core.async приносит её в Clojure.
В браузерной песочнице core.async нет, поэтому весь код урока идёт листингами и проверяется настоящим Clojure в babashka. Babashka исполняет go-блок на отдельном потоке, а не машиной состояний, поэтому вывод у листингов тот же, но разницу между <! и <!! ниже придётся понять головой, а не увидеть в профилировщике.
Канал и две руки
Канал создаёт chan. Положить в него значение и забрать значение можно двумя парами операций, и сначала возьмём ту, что попроще:
(require '[clojure.core.async :as a])
(def events (a/chan))
;; Отдельный поток кладёт событие. >!! ждёт, пока его кто-нибудь заберёт.
(a/thread
(a/>!! events {:event/type :hit :damage 3}))
;; Главный поток ждёт событие и забирает его.
(def event (a/<!! events))
(println event) ; {:event/type :hit, :damage 3}
(assert (= 3 (:damage event)))>!! значит «положить и ждать», <!! значит «забрать и ждать». Двойной восклицательный знак предупреждает: операция блокирует поток. У канала без буфера отдать значение можно только из рук в руки: тот, кто кладёт, стоит, пока не придёт тот, кто забирает, и наоборот.
Канал можно закрыть через close!. После этого положить в него ничего нельзя, а чтение возвращает nil, когда значения кончились. Поэтому nil в канал класть запрещено: он означает «канал закрыт».
go-блок: процесс без потока
Поток в JVM дорогой: мегабайт памяти на стек и переключение через операционную систему. Тысячу монстров, каждый со своим потоком, сервер выдержит с трудом. go-блок решает эту задачу так же, как горутина в Go: процесс выглядит последовательным кодом, но, когда ему нужно ждать, он снимается с потока и уступает место другому.
(require '[clojure.core.async :as a])
(def events (a/chan 10))
;; Бой: кладёт три события и закрывает канал.
(a/go
(doseq [e [{:event/type :hit :damage 3}
{:event/type :miss}
{:event/type :hit :damage 9}]]
(a/>! events e))
(a/close! events))
;; Журнал: читает, пока канал не закроется, и считает урон.
(def total
(a/go-loop [sum 0]
(if-let [e (a/<! events)]
(recur (+ sum (:damage e 0)))
sum)))
;; go возвращает канал, в который положит свой результат.
(assert (= 12 (a/<!! total)))
(println "урон за бой" 12)Внутри go пишут операции с одним восклицательным знаком: >! и <!. Они не блокируют поток, а паркуют процесс. Макрос go переписывает тело в машину состояний: каждая <! становится точкой, где процесс сохраняет свои локальные переменные и отпускает поток. Когда значение появится, процесс продолжит с того же места, возможно, уже на другом потоке.
go-loop это сокращение для (go (loop ...)), с тем же recur, что в уроке про рекурсию.
| Операция | Где можно | Что делает при ожидании |
|---|---|---|
>! и <! | только внутри go | паркует go-блок, поток свободен |
>!! и <!! | вне go: в main, в thread | блокирует поток |
offer! и poll! | где угодно | не ждёт: сразу удача или nil |
Почему внутри go нельзя блокировать
go-блоки делят небольшой пул потоков, на JVM по умолчанию восемь. Если go-блок вызывает что-то блокирующее (<!!, Thread/sleep, запрос к базе, чтение файла), он держит поток пула всё время ожидания. Восемь таких блоков, и пул занят целиком: остальные тысячи процессов стоят, хотя у них всё готово.
пул go: [поток 1: ждёт базу] [поток 2: ждёт базу] ... [поток 8: ждёт базу]
очередь: go-блок 9 ── go-блок 10 ── ... ── go-блок 1000 (готовы, но потоков нет)Для блокирующей работы есть thread: он запускает тело на отдельном потоке и, как go, возвращает канал с результатом. Правило короткое: ждёшь канал внутри go, пиши <!; ждёшь что-то другое, выноси в thread.
Буферы: что делать, когда читатель не успевает
Журнал пишет на диск и иногда отстаёт от боя. Канал с буфером держит несколько значений, и писатель не ждёт, пока буфер не заполнится. Что происходит дальше, решает вид буфера:
(require '[clojure.core.async :as a])
(defn fill [buf]
(let [c (a/chan buf)
accepted (mapv #(a/offer! c %) (range 5))]
(a/close! c)
{:accepted accepted :kept (a/<!! (a/into [] c))}))
;; Фиксированный: лишнее не принимается, писатель ждал бы.
(assert (= {:accepted [true true nil nil nil] :kept [0 1]} (fill 2)))
;; Отбрасывающий: принимает всё, новое выбрасывает.
(assert (= {:accepted [true true true true true] :kept [0 1]} (fill (a/dropping-buffer 2))))
;; Скользящий: принимает всё, выбрасывает старое.
(assert (= {:accepted [true true true true true] :kept [3 4]} (fill (a/sliding-buffer 2))))
(println "буферы ведут себя как ожидалось")offer! не ждёт и честно показывает, принял ли канал значение. Выбор буфера это решение о данных, а не о скорости:
- журналу событий терять нельзя ничего: фиксированный буфер, и пусть бой подождёт;
- полоске здоровья в интерфейсе нужно только последнее значение: скользящий буфер;
- счётчику «первые сто игроков дня» важны первые: отбрасывающий.
alts! и timeout: ждать нескольких
Игрок думает над ходом. Если за пять секунд он ничего не выбрал, ход пропускается. timeout возвращает канал, который сам закроется через заданное число миллисекунд. alts! ждёт сразу на нескольких каналах и возвращает пару: значение и канал, который ответил первым.
(require '[clojure.core.async :as a])
(defn wait-move [moves ms]
(a/go
(let [[move ch] (a/alts! [moves (a/timeout ms)])]
(if (= ch moves) move :skip))))
(def moves (a/chan))
;; Никто не походил за 50 мс.
(assert (= :skip (a/<!! (wait-move moves 50))))
;; Игрок пошёл на север через 10 мс.
(a/go (a/<! (a/timeout 10)) (a/>! moves :north))
(assert (= :north (a/<!! (wait-move moves 500))))
(println "ход принят")Это select из Go с веткой time.After, такие таймауты разбираются в уроке про паттерны конкурентности в Go. В Janet ту же роль играет ev/select из урока про цикл событий.
pipeline: параллельная обработка
events.log курса аналитика разбирают построчно: строка превращается в мапу, мусор отбрасывается. Если строк миллионы, разбор хочется пустить на несколько ядер. pipeline берёт входной канал, выходной канал, число параллельных обработчиков и трансдьюсер, а порядок значений на выходе сохраняет:
(require '[clojure.core.async :as a]
'[clojure.string :as str])
(def lines
["2026-05-14T07:38:24Z player=2979 platform=ios event=install"
""
"2026-05-14T07:44:32Z player=2979 platform=ios event=tutorial_start"
"2026-05-22T13:0"])
(defn parse-line [line]
(let [fields (into {}
(keep #(let [[k v] (str/split % #"=" 2)]
(when v [(keyword k) v])))
(str/split line #" "))]
(when (:event fields) fields)))
(def out (a/chan 10))
(a/pipeline 4 out (keep parse-line) (a/to-chan! lines))
(def parsed (a/<!! (a/into [] out)))
(assert (= ["install" "tutorial_start"] (mapv :event parsed)))
(println parsed)Тот же трансдьюсер (keep parse-line) работает и с вектором через into, и с каналом через pipeline. Канал тоже принимает трансдьюсер: (a/chan 10 (filter hit?)) пропустит только удары. Как устроены трансдьюсеры и почему они не привязаны к источнику, разберём в уроке про трансдьюсеры.
Когда core.async, а когда нет
| Задача | Чем решать |
|---|---|
| Посчитать что-то в фоне и забрать результат | future |
| Одно общее значение, которое меняют многие | атом |
| Поток событий между независимыми частями | канал |
| Ждать первого из нескольких источников или таймаута | alts! |
| Параллельная обработка с сохранением порядка | pipeline |
core.async добавляет в программу второй способ думать о времени, и это цена. Если задача решается одним future или атомом, каналы не нужны. В TypeScript похожую задачу решают потоки Effect из урока про Stream: там источник, преобразования и приёмник тоже разделены, только вместо каналов значения тянет приёмник.
Проверь себя
Задание
Дальше
Babashka, на которой проверялись листинги этого урока, заслуживает своего разговора. В следующем уроке посмотрим, где ещё живёт Clojure кроме JVM-сервиса, и научимся писать тесты на clojure.test.
домашка