You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
The core.async library supports asynchronous programming through the use of channels.
6
+
7
+
To use core.async, declare a dependency on Clojure 1.10.0 or higher and the latest core.async library:
8
+
9
+
```clojure
10
+
{:deps
11
+
{org.clojure/clojure {:mvn/version"1.12.0"}
12
+
org.clojure/core.async {:mvn/version"1.6.673"}}}
13
+
```
14
+
15
+
To start working with core.async, require the `clojure.core.async` namespace at the REPL:
16
+
17
+
```clojure
18
+
(require '[clojure.core.async :as a :refer [<!! >!! <! >!]])
19
+
```
20
+
21
+
Or include it in your namespace:
22
+
23
+
```clojure
24
+
(nsmy.ns
25
+
(:require [clojure.core.async :as a :refer [<!! >!! <! >!]]))
26
+
```
27
+
28
+
## Channels
29
+
30
+
Values are conveyed on queue-like channels. By default channels are unbuffered (0-length) - they require producer and consumer to rendezvous for the transfer of a value through the channel.
31
+
32
+
Use `chan` to make an unbuffered channel:
33
+
34
+
```clojure
35
+
(a/chan)
36
+
```
37
+
38
+
Pass a number to create a channel with a fixed buffer size:
39
+
40
+
```clojure
41
+
(a/chan10)
42
+
```
43
+
44
+
`close!` a channel to stop accepting puts. Remaining values are still available to take. Drained channels return nil on take. Nils may not be sent over a channel explicitly!
45
+
46
+
```clojure
47
+
(let [c (a/chan)]
48
+
(a/close! c))
49
+
```
50
+
51
+
Channels can also use custom buffers that have different policies for the "full" case. Two useful examples are provided in the API.
52
+
53
+
```clojure
54
+
;; Use `dropping-buffer` to drop newest values when the buffer is full:
55
+
(a/chan (a/dropping-buffer10))
56
+
57
+
;; Use `sliding-buffer` to drop oldest values when the buffer is full:
58
+
(a/chan (a/sliding-buffer10))
59
+
```
60
+
61
+
## Threads
62
+
63
+
In ordinary threads, we use `>!!` (blocking put) and `<!!` (blocking take) to communicate via channels.
64
+
65
+
```clojure
66
+
(let [c (a/chan10)]
67
+
(>!! c "hello")
68
+
(assert (="hello" (<!! c)))
69
+
(a/close! c))
70
+
```
71
+
72
+
Because these are blocking calls, if we try to put on an unbuffered channel, we will block the main thread. We can use `thread` (like `future`) to execute a body in a pool thread and return a channel with the result. Here we launch a background task to put "hello" on a channel, then read that value in the current thread.
73
+
74
+
```clojure
75
+
(let [c (a/chan)]
76
+
(a/thread (>!! c "hello"))
77
+
(assert (="hello" (<!! c)))
78
+
(a/close! c))
79
+
```
80
+
81
+
## Go Blocks and IOC Threads
82
+
83
+
The `go` macro asynchronously executes its body in a special pool of threads. Channel operations that would block will pause execution instead, blocking no threads. This mechanism encapsulates the inversion of control that is external in event/callback systems. Inside `go` blocks, we use `>!` (put) and `<!` (take).
84
+
85
+
Here we convert our prior channel example to use go blocks:
86
+
87
+
```clojure
88
+
(let [c (a/chan)]
89
+
(a/go (>! c "hello"))
90
+
(assert (="hello" (<!! (a/go (<! c)))))
91
+
(a/close! c))
92
+
```
93
+
94
+
Instead of the explicit thread and blocking call, we use a go block for the producer. The consumer uses a go block to take, then returns a result channel, from which we do a blocking take.
95
+
96
+
== Alts
97
+
98
+
One killer feature for channels over queues is the ability to wait on many channels at the same time (like a socket select). This is done with `alts!!` (ordinary threads) or `alts!` in go blocks.
99
+
100
+
We can create a background thread with alts that combines inputs on either of two channels. `alts!!` takes a set of operations to perform - either a channel to take from or a [channel value] to put and returns the value (nil for put) and channel that succeeded:
101
+
102
+
```clojure
103
+
(let [c1 (a/chan)
104
+
c2 (a/chan)]
105
+
(a/thread (whiletrue
106
+
(let [[v ch] (a/alts!! [c1 c2])]
107
+
(println"Read" v "from" ch))))
108
+
(>!! c1 "hi")
109
+
(>!! c2 "there"))
110
+
```
111
+
112
+
Prints (on stdout, possibly not visible at your repl):
113
+
114
+
```
115
+
Read hi from #object[clojure.core.async.impl.channels.ManyToManyChannel ...]
116
+
Read there from #object[clojure.core.async.impl.channels.ManyToManyChannel ...]
117
+
```
118
+
119
+
We can use alts! to do the same thing with go blocks:
120
+
121
+
```clojure
122
+
(let [c1 (a/chan)
123
+
c2 (a/chan)]
124
+
(a/go (whiletrue
125
+
(let [[v ch] (a/alts! [c1 c2])]
126
+
(println"Read" v "from" ch))))
127
+
(a/go (>! c1 "hi"))
128
+
(a/go (>! c2 "there")))
129
+
```
130
+
131
+
Since go blocks are lightweight processes not bound to threads, we can have LOTS of them! Here we create 1000 go blocks that say hi on 1000 channels. We use alts!! to read them as they're ready.
132
+
133
+
```clojure
134
+
(let [n 1000
135
+
cs (repeatedly n a/chan)
136
+
begin (System/currentTimeMillis)]
137
+
(doseq [c cs] (a/go (>! c "hi")))
138
+
(dotimes [i n]
139
+
(let [[v c] (a/alts!! cs)]
140
+
(assert (="hi" v))))
141
+
(println"Read" n "msgs in" (- (System/currentTimeMillis) begin) "ms"))
142
+
```
143
+
144
+
`timeout` creates a channel that waits for a specified ms, then closes:
<p>Values are conveyed on queue-like channels. By default channels are unbuffered (0-length) - they require producer and consumer to rendezvous for the transfer of a value through the channel.</p>
20
+
<p>Use <code>chan</code> to make an unbuffered channel:</p>
21
+
<pre><codeclass="language-clojure">(a/chan)
22
+
</code></pre>
23
+
<p>Pass a number to create a channel with a fixed buffer size:</p>
24
+
<pre><codeclass="language-clojure">(a/chan 10)
25
+
</code></pre>
26
+
<p><code>close!</code> a channel to stop accepting puts. Remaining values are still available to take. Drained channels return nil on take. Nils may not be sent over a channel explicitly!</p>
<p>Because these are blocking calls, if we try to put on an unbuffered channel, we will block the main thread. We can use <code>thread</code> (like <code>future</code>) to execute a body in a pool thread and return a channel with the result. Here we launch a background task to put “hello” on a channel, then read that value in the current thread.</p>
<h2><ahref="#go-blocks-and-ioc-threads" id="go-blocks-and-ioc-threads"></a>Go Blocks and IOC Threads</h2>
51
+
<p>The <code>go</code> macro asynchronously executes its body in a special pool of threads. Channel operations that would block will pause execution instead, blocking no threads. This mechanism encapsulates the inversion of control that is external in event/callback systems. Inside <code>go</code> blocks, we use <code>>!</code> (put) and <code><!</code> (take).</p>
52
+
<p>Here we convert our prior channel example to use go blocks:</p>
<p>Instead of the explicit thread and blocking call, we use a go block for the producer. The consumer uses a go block to take, then returns a result channel, from which we do a blocking take.</p>
59
+
<p>== Alts</p>
60
+
<p>One killer feature for channels over queues is the ability to wait on many channels at the same time (like a socket select). This is done with <code>alts!!</code> (ordinary threads) or <code>alts!</code> in go blocks.</p>
61
+
<p>We can create a background thread with alts that combines inputs on either of two channels. <code>alts!!</code> takes a set of operations to perform - either a channel to take from or a <ahref="channel value">channel value</a> to put and returns the value (nil for put) and channel that succeeded:</p>
<p>Since go blocks are lightweight processes not bound to threads, we can have LOTS of them! Here we create 1000 go blocks that say hi on 1000 channels. We use alts!! to read them as they’re ready.</p>
84
+
<pre><codeclass="language-clojure">(let [n 1000
85
+
cs (repeatedly n a/chan)
86
+
begin (System/currentTimeMillis)]
87
+
(doseq [c cs] (a/go (>! c "hi")))
88
+
(dotimes [i n]
89
+
(let [[v c] (a/alts!! cs)]
90
+
(assert (= "hi" v))))
91
+
(println "Read" n "msgs in" (- (System/currentTimeMillis) begin) "ms"))
92
+
</code></pre>
93
+
<p><code>timeout</code> creates a channel that waits for a specified ms, then closes:</p>
<p>We can combine timeout with <code>alts!</code> to do timed channel waits. Here we wait for 100 ms for a value to arrive on the channel, then give up:</p>
0 commit comments