Go 分布式学习利器（19）-- Go并发编程之 CSP(communicating sequential processes) 机制

文章目录

前言
CSP 特点
CSP代码演示
- 1. 正常流程的代码
- 2. CSP 未设置buffer 代码
- 3. 设置指定大小的channel buffer
总结

前言

CSP 这个名词大家会比较陌生，但是说到future 熟悉C++ / JAVA 线程模型的伙伴可能就会很熟悉了，通过future机制能够实现两个线程之间的数据交互，当然通过通用的mutex 和 condition_variable 也能够实现，但这一些机制的引入无形中会大大增加代码的复杂度。

举例一段C++代码:

#include <iostream>
#include <future>
#include <thread>int fib(int n)
{if (n < 3) return 1;else return fib(n-1) + fib(n-2);
}int main()
{std::future<int> f1 = std::async(std::launch::async, [](){return fib(20);});std::future<int> f2 = std::async(std::launch::async, [](){return fib(25);});std::cout << "waiting...\n";//主线程在该处阻塞，直到获取到f1,f2两个线程到合法的结果f1.wait();f2.wait();std::cout << "f1: " << f1.get() << '\n';std::cout << "f2: " << f2.get() << '\n';
}

可以看到通过future能够达到延迟获取线程执行结果，主线程和两个子线程之间通信的过程就像建立了一个管道，子线程在一端塞入数据，主线程在一端取数据，管道内没有数据，主线程可以选择等待，也可以选择返回。

相关c++ 的future机制介绍可以参考C++ 多线程编程：future异步访问类。

同样Go语言中也会有这样支持协程之间便捷自由得数据交互的机制，也就是CSP

CSP 特点

CSP支持通过channel 进行通信，即channel 支持维护一定大小的buffer（有大小限制）用作消息的存储，两个通信协程只需要向channel中放入或取出即可，无需等待
CSP 支持无指定buffer大小的channel通信，即一个协程等待结果，另一个协程负责发送，只有结果被取出来才能向channel中放入。
异步返回，调用协程执行者无需等待，即可返回

基本使用语法可以参考代码演示中的第2段和第3段代码。

CSP代码演示

1. 正常流程的代码

package groutineimport ("fmt""testing""time"
)func service() string {time.Sleep(time.Millisecond * 50)return "Service is Done"
}func otherTask() {fmt.Println("do something else ")time.Sleep(time.Millisecond * 100)fmt.Println("Task is done")
}// 顺序执行
func TestService(t *testing.T) {fmt.Println(service())otherTask()
}

这里直接就是函数的调用，执行也会按照函数的调用顺序来执行

=== RUN   TestService
Service is Done
do something else
Task is done
--- PASS: TestService (0.16s)

2. CSP 未设置buffer 代码

func service() string {time.Sleep(time.Millisecond * 50)return "Service is Done"
}func otherTask() {fmt.Println("do something else ")time.Sleep(time.Millisecond * 100)fmt.Println("Task is done")
}func AsyncService() chan string {rech := make(chan string) // 声明一个channelgo func() {ret := service()fmt.Println("resturned result")rech <- ret // 向 channel 中放数据fmt.Println("service exited")}()return rech // 返回channel
}func TestAsyncService(t *testing.T) {ret := AsyncService()otherTask()// 从channel中取数据，且会阻塞到ret内部有有效数据才返回fmt.Println(<-ret) // 假如后续还有代码执行，则channel不会阻塞，// 立即执行goroutine后续的代码，直到channle内部获取到了有效的数据才会打印// time.Sleep(time.Second * 1)
}

执行如下:

=== RUN   TestAsyncService
do something else # otherTask 函数
resturned result # AsyncService 函数
Task is done # otherTask 函数
Service is Done # Service函数, 从channel中取数据，# 此时service函数还在睡眠，并未返回结果，# 这里会阻塞，直到取到有效的Service 给channel返回结果
service exited # AsyncService 函数
--- PASS: TestAsyncService (1.11s)

假如测试代码变更为:

func TestAsyncService(t *testing.T) {ret := AsyncService()otherTask()// 从channel中取数据，且会阻塞到ret内部有有效数据才返回fmt.Println(<-ret) // 假如后续还有代码执行，则channel不会阻塞，// 立即执行goroutine后续的代码，直到channle内部获取到了有效的数据才会打印time.Sleep(time.Second * 1)
}

后两行的输出则会发生变化:

=== RUN   TestAsyncService
do something else # otherTask 函数
resturned result # AsyncService 函数
Task is done # otherTask 函数
service exited # AsyncService 函数# go routine执行完毕，不会阻塞在channel的有效数据获取处
Service is Done # Service函数 后续执行完毕返回给channel，才会打印
--- PASS: TestAsyncService (1.11s)

3. 设置指定大小的channel buffer

func AsyncService() chan string {rech := make(chan string,1 ) // 声明一个channel，并指定其buffer大小为1go func() {ret := service()fmt.Println("resturned result")rech <- ret // 向 channel 中放数据fmt.Println("service exited")}()return rech // 返回channel
}func TestAsyncService(t *testing.T) {ret := AsyncService()otherTask()fmt.Println(<-ret) // 从channel中取数据，即使是最后一段逻辑，也不会阻塞go routine的执行
}

最后的输出如下:

=== RUN   TestAsyncService
do something else
resturned result
service exited
Task is done
Service is Done
--- PASS: TestAsyncService (0.10s)

可以看到有了buffer的channel 不会阻塞go routine的执行，后续有了有效数据，直接从channel中取就可以了。

总结

Go 的CSP并发机制比较烧脑，协程之间的数据通信可以异步获取结果而不需要等待正常逻辑的完成。
建议大家将以上代码本地多多调试一番，对于CSP机制和应用场景的理解会更加深刻一些。