js-csp 异步编程的一些简单的例子

时间  2019-12-05
标签 csp 异步 编程 一些 简单 例子 栏目 JavaScript 繁體版
原文   原文链接

发现一个月没刷技术文章了, 有点慌, 整理一篇短的 CSP 用法出来,
只包含最基本的用法, 在 Go 里边最清晰, 不过我是在 Clojure 写的 CSP.
js 版本的 CSP 实现包含异步, 用法会更繁琐一些, 可是也值得看看.
我相信 async/await 普及以前, js-csp 仍是一个颇有意思的选择.git

个人代码写的是 CoffeeScript, 能够自动脑补圆括号花括号上去...
注意包含 yield 的函数自动被转成 function*() {}, 因此注意脑补.
脑补不出来的只好贴在这边编译下了 http://coffeescript.org/github

使用 timeout

首先是最基本的 CSP 的例子, 也就是用同步的代码写异步的逻辑,
CSP 当中最核心的概念是 Channel, 最简单的 csp.timeout(1000) 建立 channel.数组

csp = require 'js-csp'

# 用 csp.go 启动一个 yield 函数
csp.go ->
  # 有 csp.take 从这个管道取出数据, yield 来模拟阻塞的效果
  yield csp.take csp.timeout(1000)
  console.log 'Gone 1s'

运行一下:缓存

=>> coffee async.coffee
Gone 1s

我注意到对于 timeout 来讲, 省掉 csp.take 也是可以正常运行的:dom

csp = require 'js-csp'

csp.go -> # 脑补 yield 函数

  yield csp.timeout 1000
  console.log 'Gone 1s'
  yield csp.timeout 2000
  console.log 'Gone 2s'
  yield csp.timeout 3000
  console.log 'Gone 3s. End'

运行一下:异步

=>> coffee async.coffee
Gone 1s
Gone 2s
Gone 3s. End

使用 put 和 take

csp.timeout 比较特殊, 默认就会产生数据, 只要进行 csp.take 就行了.
通常的 Channel 的话, 须要手动建立出来, 而后手动推数据,
好比下面的代码建立了一个数据, 用 csp.go 启动另外一个"进程"往 Channel 推数据,
这里的"进程"的说法并非真正的进程, 只是模拟进程的行为:async

csp = require 'js-csp'

talk = (ch) ->
  yield csp.timeout 3000
  console.log 'Done 3s timeout'
  # 等待 3s 而后往 Channel 当中写入数据, yield 会产生等待
  yield csp.put ch, 'some result'

csp.go ->
  ch = csp.chan()

  # 启动另外一个"进程"
  csp.go talk, [ch] # 数组里是传给 talk 函数的参数

  # 使用 yield.take 从 Channel 取出数据, 使用 yield 模拟等待
  result = yield csp.take ch
  console.log 'Result:', JSON.stringify(result)

运行一下:函数

=>> coffee async.coffee
Done 3s timeout
Result: "some result"

伪装有两个进程

一样是上边的代码, 只是调整一下写法, 看上去像是分别启动了两个"进程",
虽然它们的运行时独立的, 可是能够经过管道进行通讯,
并且在对应的 csp.take 和 csp.put 操做过程当中, 会经过 yield 进行等待:oop

csp = require 'js-csp'

talk = (ch) ->
  yield csp.timeout 3000
  console.log 'Done 3s timeout'
  yield csp.put ch, 'some result'

listen = (ch) ->
  result = yield csp.take ch
  console.log 'Result:', JSON.stringify(result)

# 建立 Channel, 启动两个"进程"
theCh = csp.chan()
# csp.go 后面第一个是 yield 函数, 第二个是参数的数组, 虽然比较难看
csp.go talk, [theCh]
csp.go listen, [theCh]

运行一下:ui

=>> coffee async.coffee
Done 3s timeout
Result: "some result"

封装异步事件

实际使用当中, 会须要把 js 环境的异步代码封装成管道的形式,
不封装成管道, 就不能借助 csp.go 来封装同步代码了,
因为 js 不像 Go 那样整个语言层面作了处理, 实际上会有奇怪的写法,
因此 js-csp 提供了 csp.putAsync 和 csp.takeAsync:

csp = require 'js-csp'

talk = (ch) ->
  setTimeout ->
    csp.putAsync ch, 'some result'
    console.log 'Finished 3s of async'
  , 3000

listen = (ch) ->
  result = yield csp.take ch
  console.log 'Result:', JSON.stringify(result)

theCh = csp.chan()
talk theCh
csp.go listen, [theCh]

运行一下:

=>> coffee async.coffee
Finished 3s of async
Result: "some result"

处理超时

一个操做是否超时的问题, 能够同时启动一个定时的"进程",
而后观察两个"进程"哪个先执行完成, 从而判断是否超时,
这就用到了 csp.alts 函数, 这个奇怪的命名是用 Clojure 带过来的:

csp = require 'js-csp'

talk = (ch) ->
  time = Math.random() * 4 * 1000
  setTimeout ->
    console.log "Get result after #{time}ms"
    csp.putAsync ch, 'some result'
  , time

listen = (ch) ->
  hurry = csp.timeout 2000
  # 经过 csp.alts 同时等待多个 Channel 返回数据
  result = yield csp.alts [ch, hurry]
  # result.channel 能够用于判断数据的来源, result.value 才是真正的数据
  if result.channel is hurry
    console.log 'Too slow, got no result'
    # close 只是设置 Channel 的状态, 其实还须要手工处理一些逻辑
    hurry.close()
  else
    console.log 'Fast enough, got', JSON.stringify(result.value)

theCh = csp.chan()
talk theCh
csp.go listen, [theCh]

用了随机数, 运行屡次试一下, 能够看到根据不一样的时间, 结果是不同的:

=>> coffee async.coffee
Too slow, got no result
Get result after 3503.6168682995008ms

=>> coffee async.coffee
Too slow, got no result
Get result after 3095.264637685924ms

=>> coffee async.coffee
Get result after 703.6501633183257ms
Fast enough, got "some result"

=>> coffee async.coffee
Too slow, got no result
Get result after 3729.5125755664317ms

=>> coffee async.coffee
Get result after 101.51519531067788ms
Fast enough, got "some result"

循环任务

跟 yield 用法相似, 若是有循环的代码, 也能够用 CSP 写出来,
这个的话不用怎么想应该能明白了, loop 只是 while true 的语法糖:

csp = require 'js-csp'

chatter = (ch) ->
  counter = 0
  loop
    yield csp.timeout 1000
    counter += 1
    yield csp.put ch, counter

repeat = (ch) ->
  loop
    something = yield csp.take ch
    console.log 'Hear something:', something

theCh = csp.chan()
csp.go chatter, [theCh]
csp.go repeat, [theCh]

运行一下:

=>> coffee async.coffee
Hear something: 1
Hear something: 2
Hear something: 3
Hear something: 4
^C

多个数据的消费者

实际场景当中会遇到多个消费者从单个生产者读取数据的需求,
这是一个用 Channel 比较合适的场景, 启动两个"进程"读取一个 Channel 就行了,
下面我模拟的是不一样的处理时间 300ms 和 800ms 读取 100ms 频率的数据,
由于 CSP 自动处理了等待, 整个代码看上去挺简单的:

csp = require 'js-csp'

chatter = (ch) ->
  counter = 0
  loop
    yield csp.timeout 100
    counter += 1
    yield csp.put ch, counter

repeat = (ch) ->
  loop
    yield csp.timeout 800
    something = yield csp.take ch
    console.log 'Hear at 1:', something

repeat2 = (ch) ->
  loop
    yield csp.timeout 300
    something = yield csp.take ch
    console.log 'Hear at 2:', something

theCh = csp.chan()
csp.go chatter, [theCh]
csp.go repeat, [theCh]
csp.go repeat2, [theCh]

运行一下:

=>> coffee async.coffee
Hear at 2: 1
Hear at 2: 2
Hear at 1: 3
Hear at 2: 4
Hear at 2: 5
Hear at 2: 6
Hear at 1: 7
Hear at 2: 8
Hear at 2: 9
Hear at 1: 10
Hear at 2: 11
Hear at 2: 12
Hear at 2: 13
Hear at 1: 14
Hear at 2: 15
Hear at 2: 16
Hear at 1: 17
Hear at 2: 18
Hear at 2: 19
Hear at 2: 20
Hear at 1: 21
Hear at 2: 22
Hear at 2: 23
Hear at 1: 24
^C

使用 buffer

默认状况下管道是阻塞的, csp.put csp.take 成对进行,
也就是说, 只有一个就绪的话, 它会等待另外一个开始, 而后一块儿执行,
可是用 buffer 的话, 管道就会先在必定范围内进行缓存,
这样 csp.put 就能够先运行下去了, 这个是不难理解的...
管道实际上有 3 种策略, fixed, dropping, sliding:

  • fixed, 缓存放满之后就会开始造成阻塞了

  • dropping, 缓存满了之后, 新的数据就会丢弃

  • sliding, 缓存满之后, 会丢弃掉旧的数据让新数据能放进缓存

随便演示一个丢弃数据的例子:

csp = require 'js-csp'

chatter = (ch) ->
  counter = 0
  loop
    yield csp.timeout 200
    counter += 1
    console.log 'Write data:', counter
    yield csp.put ch, counter

repeat = (ch) ->
  loop
    yield csp.timeout 300
    something = yield csp.take ch
    console.log 'Hear:', something

theCh = csp.chan(csp.buffers.dropping(3))
csp.go chatter, [theCh]
csp.go repeat, [theCh]

运行一下, 能够看到 "Hear" 部分丢失了一些数据, 但前三个数据不会丢:

=>> coffee async.coffee
Write data: 1
Hear: 1
Write data: 2
Hear: 2
Write data: 3
Write data: 4
Hear: 3
Write data: 5
Hear: 4
Write data: 6
Write data: 7
Hear: 5
Write data: 8
Hear: 6
Write data: 9
Write data: 10
Hear: 7
Write data: 11
Hear: 8
Write data: 12
Write data: 13
Hear: 9
Write data: 14
Hear: 11
Write data: 15
Write data: 16
Hear: 12
Write data: 17
Hear: 14
^C

小结

因为 CSP 是在 Go 语言发明的, 完整的用法仍是看 Go 的教程比较好,
到了 Clojure 和 js 当中不免会增长一些坑, 特别是 js 当中...
上面提到的 API 在 js-csp 的文档上有描述, 例子也有, 可是挺少的:

另外还有一些高级一点的用法, 好比数据的 transform 和 pipe 之类的,其实就是 Stream 的用法在 Channel 上的改版, 某种程度上 Channel 也是 Stream,对于我我的来讲, Channel 的抽象比起 Stream 的抽象舒服多了.

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程