闭包代码如何控制高并发的正确

关于处理闭包的并发问题#

闭包(closure)的概念在很多语言中都有。闭包通常在函数式编程语言或者具有函数式特性的编程语言中会单独列出来，作为一个语言特性。以展示这个语言的强大。

什么是闭包 我们通常对闭包的解释是带有运行环境上下文变量的函数

给一个闭包的代码示例(Go)

1
a := func() func() {
2
  i := 0
3
  return func() {
4
    i++
5
    fmt.Println(i)
6
  }
7
}()

上面的这个函数就实现了闭包，每调用一次，内部的变量i就增加1。而且这个变量从外部访问不到

简单的闭包我们都能写，问题是我们如果处理高并发场景下的闭包问题呢，例如下面这段代码

1
// 原子计数器
2
ato := int32(0)
3
// 闭包
4
a := func() func() {
5
  i := 0
6
  return func() {
7
    atomic.AddInt32(&ato, 1)
8
    i++
9
    if ato == 1000 {
10
      fmt.Println("i -> ", i, "ato -> ", ato)
11
    }
12
  }
13
}()
14
// 记录调用了多少次协程执行
15
for n := 0; n < 1000; n++ {
16
  go a()
17
}
18
// 等待所有协程执行完毕
19
for ato < int32(1000) {
20
  time.Sleep(1000000)
21
}
22
fmt.Println("done")

我们期望的输出值是1000，但是i的实际的输出值有可能不到1000。

原因也很简单，因为i++并不是原子性的。由于协程运行的乱序执行，导致有可能会出现两次以3 为基数的自增，这个时候，两次自增的结果都是4就有一次自增相当于无效化

所以我们处理的方法也可以从i++的原子性来考虑

方法一#

使用原子操作#

1
ch := make(chan int32, j)
2
a := func() func() {
3
  i := int32(0)
4
  ch <- i
5
  return func() {
6
    atomic.AddInt32(&i, 1) // 原子操作
7
    fmt.Println(i)
8
  }
9
}()

这种情况下， i的值的更新都变成了原子操作，即便乱序执行，结果也是一致的

原子类的局限#

如果我们要锁定的不是一个变量的变化，而是一系列的代码操作，这个时候原子类就无用武之地了

这种时候要么用锁，要么用下面的处理办法

方法二#

使用csp机制来控制并发。csp是基于消息的常用并发模型的一种（另一种是ActorModel）。

csp传递值变量#

go语言中有一个csp并发模型，我们可以把要更新的值放到channel中，使用的时候取出来更新完在放回去，由于channel的阻塞性，自然就实现了一致性保证。我们用go来写个代码测试一下

1
a := func() func() {
2
    i := 0
3
    ch := make(chan int, 1)
4
    ch <- i
5
    return func() {
6
        t := <-ch
7
        t++
8
        fmt.Printf(" i %d, j %d\n", t, j)
9
        ch <- t
10
    }
11
}()

channel传递函数指针#

或者更激进一点，把整个函数指针放到channel中，这样就能对整个闭包函数中的代码加锁了。

1
a := func() func() {
2
  i := 0
3
  j := 0
4
  return func() {
5
    i++
6
    j++
7
    fmt.Println("i ->", i, " j ->", j)
8
  }
9
}()
10

11
fc := make(chan func(), 1)
12
fc <- a // 先预置一个用于启动
13
for p := 0; p < j; p++ {
14
  t := <-fc
15
  t()
16
  fc <- a
17
}
18
m := <-fc // 检查最终的结果
19
m()

这种方法对语言有要求，需要语言对函数支持比较好，像java这种oop语言就没法使用

rust的channel实现#

rust里面也可以跟go一样使用channel 传递函数指针来实现闭包函数的并发控制

1
use std::sync::mpsc::channel;
2

3
fn c() -> Box<dyn FnMut() -> i32> {
4
    let mut a = 1;
5
    Box::new(move || {
6
        a += 1;
7
        println!("a -> {}", a);
8
        a
9
    })
10
}
11
fn main() {
12
    let mut t = c();
13
    t();
14
    t();
15
    let (s, r) = channel::<Box<dyn FnMut() -> i32>>();
16
    if let (Ok(_), Ok(mut n)) = (s.send(t), r.recv()) {
17
        n();
18
    }
19
}