仅用大约 65 行代码，开发一个用于生成随机数、支持并发的 TCP 服务端。TCP 和 UDP 服务端随处可见，它们基于 TCP/IP 协议栈，通过网络为客户端提供服务。在这篇文章中，我将介绍如何使用 Go 语言 开发一个用于返回随机数、支持并发的 TCP 服务端。对于每一个来自 TCP 客户端的连接，它都会启动一个新的 goroutine（轻量级线程）来处理相应的请求。你可以在 GitHub 上找到本项目的源码：concTcp.go。处理 TCP 连接这个程序的主要逻辑在 handleConnection() 函数中，具体实现如下：func handleConnection(c net.Conn) { fmt.Printf("Serving %s\n", c.RemoteAddr().String()) for { netData, err := bufio.NewReader(c).ReadString('\n') if err != nil { fmt.Println(err) return } temp := strings.TrimSpace(string(netData)) if temp == "STOP" { break } result := strconv.Itoa(random()) + "\n" c.Write([]byte(string(result))) } c.Close() } 如果 TCP 客户端发送了一个 “STOP” 字符串，为它提供服务的 goroutine 就会终止；否则，TCP 服务端就会返回一个随机数给它。只要客户端不主动终止，服务端就会一直提供服务，这是由 for 循环保证的。具体来说，for 循环中的代码使用了 bufio.NewReader(c).ReadString('\n') 来逐行读取客户端发来的数据，并使用 c.Write([]byte(string(result))) 来返回数据（生成的随机数）。你可以在 Go 的 net 标准包 文档 中了解更多。支持并发在 main() 函数的实现部分，每当 TCP 服务端收到 TCP 客户端的连接请求，它都会启动一个新的 goroutine 来为这个请求提供服务。func main() { arguments := os.Args if len(arguments) == 1 { fmt.Println("Please provide a port number!") return } PORT := ":" + arguments[1] l, err := net.Listen("tcp4", PORT) if err != nil { fmt.Println(err) return } defer l.Close() rand.Seed(time.Now().Unix()) for { c, err := l.Accept() if err != nil { fmt.Println(err) return } go handleConnection(c) } } 首先，main() 确保程序至少有一个命令行参数。注意，现有代码并没有检查这个参数是否为有效的 TCP 端口号。不过，如果它是一个无效的 TCP 端口号，net.Listen() 就会调用失败，并返回一个错误信息，类似下面这样：$ go run concTCP.go 12a listen tcp4: lookup tcp4/12a: nodename nor servname provided, or not known $ go run concTCP.go -10 listen tcp4: address -10: invalid port net.Listen() 函数用于告诉 Go 接受网络连接，因而承担了服务端的角色。它的返回值类型是 net.Conn，后者实现了 io.Reader 和 io.Writer 接口。此外，main() 函数中还调用了 rand.Seed() 函数，用于初始化随机数生成器。最后，for 循环允许程序一直使用 Accept() 函数来接受 TCP 客户端的连接请求，并以 goroutine 的方式来运行 handleConnection(c) 函数，处理客户端的后续请求。net.Listen() 的第一个参数net.Listen() 函数的第一个参数定义了使用的网络类型，而第二个参数定义了服务端监听的地址和端口号。第一个参数的有效值为 tcp、tcp4、tcp6、udp、udp4、udp6、ip、ip4、ip6、Unix（Unix 套接字）、Unixgram 和 Unixpacket，其中：tcp4、udp4 和 ip4 只接受 IPv4 地址，而 tcp6、udp6 和 ip6 只接受 IPv6 地址。服务端并发测试concTCP.go 需要一个命令行参数，来指定监听的端口号。当它开始服务 TCP 客户端时，你会得到类似下面的输出：$ go run concTCP.go 8001 Serving 127.0.0.1:62554 Serving 127.0.0.1:62556 netstat 的输出可以确认 congTCP.go 正在为多个 TCP 客户端提供服务，并且仍在继续监听建立连接的请求：$ netstat -anp TCP | grep 8001 tcp4 0 0 127.0.0.1.8001 127.0.0.1.62556 ESTABLISHED tcp4 0 0 127.0.0.1.62556 127.0.0.1.8001 ESTABLISHED tcp4 0 0 127.0.0.1.8001 127.0.0.1.62554 ESTABLISHED tcp4 0 0 127.0.0.1.62554 127.0.0.1.8001 ESTABLISHED tcp4 0 0 *.8001 *.* LISTEN 在上面输出中，最后一行显示了有一个进程正在监听 8001 端口，这意味着你可以继续连接 TCP 的 8001 端口。第一行和第二行显示了有一个已建立的 TCP 网络连接，它占用了 8001 和 62556 端口。相似地，第三行和第四行显示了有另一个已建立的 TCP 连接，它占用了 8001 和 62554 端口。下面这张图片显示了 concTCP.go 在服务多个 TCP 客户端时的输出：类似地，下面这张图片显示了两个 TCP 客户端的输出（使用了 nc 工具）：你可以在 维基百科 上找到更多关于 nc（即 netcat）的信息。总结现在，你学会了如何用大约 65 行 Go 代码来开发一个生成随机数、支持并发的 TCP 服务端，这真是太棒了！如果你想要让你的 TCP 服务端执行别的任务，只需要修改 handleConnection() 函数即可。via: https://opensource.com/article/18/5/building-concurrent-tcp-server-go作者：Mihalis Tsoukalos 选题：lkxed 译者：lkxed 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
通过学习如何定位并发处理的陷阱来避免未来处理这些问题时的困境。在复杂的分布式系统进行任务处理时，你通常会需要进行并发的操作。在 Mode.net 公司，我们每天都要和实时、快速和灵活的软件打交道。而没有一个高度并发的系统，就不可能构建一个毫秒级的动态地路由数据包的全球专用网络。这个动态路由是基于网络状态的，尽管这个过程需要考虑众多因素，但我们的重点是链路指标。在我们的环境中，链路指标可以是任何跟网络链接的状态和当前属性（如链接延迟）有关的任何内容。并发探测链接监控我们的动态路由算法 H.A.L.O.（ 逐跳自适应链路状态最佳路由 Hop-by-Hop Adaptive Link-State Optimal Routing ）部分依赖于链路指标来计算路由表。这些指标由位于每个 PoP（ 存活节点 Point of Presence ）上的独立组件收集。PoP 是表示我们的网络中单个路由实体的机器，通过链路连接并分布在我们的网络拓扑中的各个位置。某个组件使用网络数据包探测周围的机器，周围的机器回复数据包给前者。从接收到的探测包中可以获得链路延迟。由于每个 PoP 都有不止一个临近节点，所以这种探测任务实质上是并发的：我们需要实时测量每个临近连接点的延迟。我们不能串行地处理；为了计算这个指标，必须尽快处理每个探测。序列号和重置：一个重新排列场景我们的探测组件互相发送和接收数据包，并依靠序列号进行数据包处理。这旨在避免处理重复的包或顺序被打乱的包。我们的第一个实现依靠特殊的序列号 0 来重置序列号。这个数字仅在组件初始化时使用。主要的问题是我们考虑了递增的序列号总是从 0 开始。在该组件重启后，包的顺序可能会重新排列，某个包的序列号可能会轻易地被替换成重置之前使用过的值。这意味着，后继的包都会被忽略掉，直到排到重置之前用到的序列值。UDP 握手和有限状态机这里的问题是该组件重启前后的序列号是否一致。有几种方法可以解决这个问题，经过讨论，我们选择了实现一个带有清晰状态定义的三步握手协议。这个握手过程在初始化时通过链接建立会话。这样可以确保节点通过同一个会话进行通信且使用了适当的序列号。为了正确实现这个过程，我们必须定义一个有清晰状态和过渡的有限状态机。这样我们就可以正确管理握手过程中的所有极端情况。会话 ID 由握手的初始化程序生成。一个完整的交换顺序如下：发送者发送一个 SYN(ID) 数据包。接收者存储接收到的 ID 并发送一个 SYN-ACK(ID)。发送者接收到 SYN-ACK(ID) 并发送一个 ACK(ID)。它还发送一个从序列号 0 开始的数据包。接收者检查最后接收到的 ID，如果 ID 匹配，则接受 ACK(ID)。它还开始接受序列号为 0 的数据包。处理状态超时基本上，每种状态下你都需要处理最多三种类型的事件：链接事件、数据包事件和超时事件。这些事件会并发地出现，因此你必须正确处理并发。链接事件包括网络连接或网络断开的变化，相应的初始化一个链接会话或断开一个已建立的会话。数据包事件是控制数据包（SYN/SYN-ACK/ACK）或只是探测响应。超时事件在当前会话状态的预定超时时间到期后触发。这里面临的最主要的问题是如何处理并发的超时到期和其他事件。这里很容易陷入死锁和资源竞争的陷阱。第一种方法本项目使用的语言是 Golang。它确实提供了原生的同步机制，如自带的通道和锁，并且能够使用轻量级线程来进行并发处理。gopher 们聚众狂欢首先，你可以设计两个分别表示我们的会话和超时处理程序的结构体。type Session struct { State SessionState Id SessionId RemoteIp string } type TimeoutHandler struct { callback func(Session) session Session duration int timer *timer.Timer }Session 标识连接会话，内有表示会话 ID、临近的连接点的 IP 和当前会话状态的字段。TimeoutHandler 包含回调函数、对应的会话、持续时间和指向调度计时器的指针。每一个临近连接点的会话都包含一个保存调度 TimeoutHandler 的全局映射。SessionTimeout map[Session]*TimeoutHandler下面方法注册和取消超时：// schedules the timeout callback function. func (timeout* TimeoutHandler) Register() { timeout.timer = time.AfterFunc(time.Duration(timeout.duration) * time.Second, func() { timeout.callback(timeout.session) }) } func (timeout* TimeoutHandler) Cancel() { if timeout.timer == nil { return } timeout.timer.Stop() }你可以使用类似下面的方法来创建和存储超时：func CreateTimeoutHandler(callback func(Session), session Session, duration int) *TimeoutHandler { if sessionTimeout[session] == nil { sessionTimeout[session] := new(TimeoutHandler) } timeout = sessionTimeout[session] timeout.session = session timeout.callback = callback timeout.duration = duration return timeout }超时处理程序创建后，会在经过了设置的 duration 时间（秒）后执行回调函数。然而，有些事件会使你重新调度一个超时处理程序（与 SYN 状态时的处理一样，每 3 秒一次）。为此，你可以让回调函数重新调度一次超时：func synCallback(session Session) { sendSynPacket(session) // reschedules the same callback. newTimeout := NewTimeoutHandler(synCallback, session, SYN_TIMEOUT_DURATION) newTimeout.Register() sessionTimeout[state] = newTimeout }这次回调在新的超时处理程序中重新调度自己，并更新全局映射 sessionTimeout。数据竞争和引用你的解决方案已经有了。可以通过检查计时器到期后超时回调是否执行来进行一个简单的测试。为此，注册一个超时，休眠 duration 秒，然后检查是否执行了回调的处理。执行这个测试后，最好取消预定的超时时间（因为它会重新调度），这样才不会在下次测试时产生副作用。令人惊讶的是，这个简单的测试发现了这个解决方案中的一个问题。使用 cancel 方法来取消超时并没有正确处理。以下顺序的事件会导致数据资源竞争：你有一个已调度的超时处理程序。线程 1：你接收到一个控制数据包，现在你要取消已注册的超时并切换到下一个会话状态（如发送 SYN 后接收到一个 SYN-ACK）你调用了 timeout.Cancel()，这个函数调用了 timer.Stop()。（请注意，Golang 计时器的停止不会终止一个已过期的计时器。）线程 2：在取消调用之前，计时器已过期，回调即将执行。执行回调，它调度一次新的超时并更新全局映射。线程 1：切换到新的会话状态并注册新的超时，更新全局映射。两个线程并发地更新超时映射。最终结果是你无法取消注册的超时，然后你也会丢失对线程 2 重新调度的超时的引用。这导致处理程序在一段时间内持续执行和重新调度，出现非预期行为。锁也解决不了问题使用锁也不能完全解决问题。如果你在处理所有事件和执行回调之前加锁，它仍然不能阻止一个过期的回调运行：func (timeout* TimeoutHandler) Register() { timeout.timer = time.AfterFunc(time.Duration(timeout.duration) * time._Second_, func() { stateLock.Lock() defer stateLock.Unlock() timeout.callback(timeout.session) }) }现在的区别就是全局映射的更新是同步的，但是这还是不能阻止在你调用 timeout.Cancel() 后回调的执行 —— 这种情况出现在调度计时器过期了但是还没有拿到锁的时候。你还是会丢失一个已注册的超时的引用。使用取消通道你可以使用取消通道，而不必依赖不能阻止到期的计时器执行的 golang 函数 timer.Stop()。这是一个略有不同的方法。现在你可以不用再通过回调进行递归地重新调度；而是注册一个死循环，这个循环接收到取消信号或超时事件时终止。新的 Register() 产生一个新的 go 线程，这个线程在超时后执行你的回调，并在前一个超时执行后调度新的超时。返回给调用方一个取消通道，用来控制循环的终止。func (timeout *TimeoutHandler) Register() chan struct{} { cancelChan := make(chan struct{}) go func () { select { case _ = <- cancelChan: return case _ = <- time.AfterFunc(time.Duration(timeout.duration) * time.Second): func () { stateLock.Lock() defer stateLock.Unlock() timeout.callback(timeout.session) } () } } () return cancelChan } func (timeout* TimeoutHandler) Cancel() { if timeout.cancelChan == nil { return } timeout.cancelChan <- struct{}{} }这个方法给你注册的所有超时提供了取消通道。一个取消调用向通道发送一个空结构体并触发取消操作。然而，这并不能解决前面的问题；可能在你通过通道取消之前以及超时线程拿到锁之前，超时时间就已经到了。这里的解决方案是，在拿到锁之后，检查一下超时范围内的取消通道。 case _ = <- time.AfterFunc(time.Duration(timeout.duration) * time.Second): func () { stateLock.Lock() defer stateLock.Unlock() select { case _ = <- handler.cancelChan: return default: timeout.callback(timeout.session) } } () }最终，这可以确保在拿到锁之后执行回调，不会触发取消操作。小心死锁这个解决方案看起来有效；但是还是有个隐患：死锁。请阅读上面的代码，试着自己找到它。考虑下描述的所有函数的并发调用。这里的问题在取消通道本身。我们创建的是无缓冲通道，即发送的是阻塞调用。当你在一个超时处理程序中调用取消函数时，只有在该处理程序被取消后才能继续处理。问题出现在，当你有多个调用请求到同一个取消通道时，这时一个取消请求只被处理一次。当多个事件同时取消同一个超时处理程序时，如连接断开或控制包事件，很容易出现这种情况。这会导致死锁，可能会使应用程序停机。有人在听吗？（已获得 Trevor Forrey 授权。）这里的解决方案是创建通道时指定缓存大小至少为 1，这样向通道发送数据就不会阻塞，也显式地使发送变成非阻塞的，避免了并发调用。这样可以确保取消操作只发送一次，并且不会阻塞后续的取消调用。func (timeout* TimeoutHandler) Cancel() { if timeout.cancelChan == nil { return } select { case timeout.cancelChan <- struct{}{}: default: // can’t send on the channel, someone has already requested the cancellation. } }总结在实践中你学到了并发操作时出现的常见错误。由于其不确定性，即使进行大量的测试，也不容易发现这些问题。下面是我们在最初的实现中遇到的三个主要问题：在非同步的情况下更新共享数据这似乎是个很明显的问题，但如果并发更新发生在不同的位置，就很难发现。结果就是数据竞争，由于一个更新会覆盖另一个，因此对同一数据的多次更新中会有某些更新丢失。在我们的案例中，我们是在同时更新同一个共享映射里的调度超时引用。（有趣的是，如果 Go 检测到在同一个映射对象上的并发读写，会抛出致命错误 — 你可以尝试下运行 Go 的数据竞争检测器）。这最终会导致丢失超时引用，且无法取消给定的超时。当有必要时，永远不要忘记使用锁。不要忘记同步 gopher 们的工作缺少条件检查在不能仅依赖锁的独占性的情况下，就需要进行条件检查。我们遇到的场景稍微有点不一样，但是核心思想跟条件变量是一样的。假设有个一个生产者和多个消费者使用一个共享队列的经典场景，生产者可以将一个元素添加到队列并唤醒所有消费者。这个唤醒调用意味着队列中的数据是可访问的，并且由于队列是共享的，消费者必须通过锁来进行同步访问。每个消费者都可能拿到锁；然而，你仍然需要检查队列中是否有元素。因为在你拿到锁的瞬间并不知道队列的状态，所以还是需要进行条件检查。在我们的例子中，超时处理程序收到了计时器到期时发出的“唤醒”调用，但是它仍需要检查是否已向其发送了取消信号，然后才能继续执行回调。如果你要唤醒多个 gopher，可能就需要进行条件检查死锁当一个线程被卡住，无限期地等待一个唤醒信号，但是这个信号永远不会到达时，就会发生这种情况。死锁可以通过让你的整个程序停机来彻底杀死你的应用。在我们的案例中，这种情况的发生是由于多次发送请求到一个非缓冲且阻塞的通道。这意味着向通道发送数据只有在从这个通道接收完数据后才能返回。我们的超时线程循环迅速从取消通道接收信号；然而，在接收到第一个信号后，它将跳出循环，并且再也不会从这个通道读取数据。其他的调用会一直被卡住。为避免这种情况，你需要仔细检查代码，谨慎处理阻塞调用，并确保不会发生线程饥饿。我们例子中的解决方法是使取消调用成为非阻塞调用 — 我们不需要阻塞调用。via: https://opensource.com/article/19/12/go-common-pitfalls作者：Eduardo Ferreira 选题：lujun9972 译者：lxbwolf 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
Pony，一种“Rust 遇上 Erlang”的语言，让开发快捷、安全、高效、高并发的程序更简单。在 Wallaroo Labs，我是工程副总裁，我们正在构建一个用 Pony 编程语言编写的 高性能分布式流处理器。大多数人没有听说过 Pony，但它一直是 Wallaroo 的最佳选择，它也可能成为你的下一个项目的最佳选择。“一门编程语言只是另一种工具。与语法无关，与表达性无关，与范式或模型无关，仅与解决难题有关。” —Sylvan Clebsch，Pony 的创建者我是 Pony 项目的贡献者，但在这里我要谈谈为什么 Pony 对于像 Wallaroo 这样的应用是个好选择，并分享我使用 Pony 的方式。如果你对我们为什么使用 Pony 来编写 Wallaroo 甚感兴趣，我们有一篇关于它的 博文。Pony 是什么？你可以把 Pony 想象成某种“Rust 遇上 Erlang”的东西。Pony 有着最引人注目的特性，它们是：类型安全存储安全异常安全无数据竞争无死锁此外，它可以被编译为高效的本地代码，它是在开放的情况下开发的，在两句版 BSD 许可证下发布。以上说的功能不少，但在这里我将重点关注那些对我们公司来说采用 Pony 至关重要的功能。为什么使用 Pony？使用大多数我们现有的工具编写快速、安全、高效、高并发的程序并非易事。“快速、高效、高并发”是可实现的目标，但加入“安全”之后，就困难了许多。对于 Wallaroo，我们希望同时实现四个目标，而 Pony 让实现它们更加简单。高并发Pony 让并发变得简单。部分是通过提供一个固执的并发方式实现的。在 Pony 语言中，所有的并发都是通过 Actor 模型 进行的。Actor 模型以在 Erlang 和 Akka 中的实现最为著名。Actor 模型出现于上世纪 70 年代，细节因实现方式而异。不变的是，所有计算都由通过异步消息进行通信的 actor 来执行。你可以用这种方式来看待 Actor 模型：面向对象中的对象是状态 + 同步方法，而 actor 是状态 + 异步方法。当一个 actor 收到一个消息时，它执行相应的方法。该方法可以在只有该 actor 可访问的状态下运行。Actor 模型允许我们以并发安全的方式使用可变状态。每个 actor 都是单线程的。一个 actor 中的两个方法绝不会并发运行。这意味着，在给定的 actor 中，数据更新不会引起数据竞争或通常与线程和可变状态相关的其他问题。快速高效Pony actor 通过一个高效的工作窃取调度程序来调度。每个可用的 CPU 都有一个单独 Pony 调度程序。这种每个核心一个线程的并发模型是 Pony 尝试与 CPU 协同工作以尽可能高效运行的一部分。Pony 运行时尝试尽可能利用 CPU 缓存。代码越少干扰缓存，运行得越好。Pony 意在帮你的代码与 CPU 缓存友好相处。Pony 的运行时还会有每个 actor 的堆，因此在垃圾收集期间，没有 “停止一切” 的垃圾收集步骤。这意味着你的程序总是至少能做一点工作。因此 Pony 程序最终具有非常一致的性能和可预测的延迟。安全Pony 类型系统引入了一个新概念：引用能力，它使得数据安全成为类型系统的一部分。Pony 语言中每种变量的类型都包含了有关如何在 actor 之间分享数据的信息。Pony 编译器用这些信息来确认，在编译时，你的代码是无数据竞争和无死锁的。如果这听起来有点像 Rust，那是因为本来就是这样的。Pony 的引用功能和 Rust 的借用检查器都提供数据安全性；它们只是以不同的方式来接近这个目标，并有不同的权衡。Pony 适合你吗？决定是否要在一个非业余爱好的项目上使用一门新的编程语言是困难的。与其他方法想比，你必须权衡工具的适当性和不成熟度。那么，Pony 和你搭不搭呢？如果你有一个困难的并发问题需要解决，那么 Pony 可能是一个好选择。解决并发应用问题是 Pony 之所以存在的理由。如果你能用一个单线程的 Python 脚本就完成所需操作，那你大概不需要它。如果你有一个困难的并发问题，你应该考虑 Pony 及其强大的无数据竞争、并发感知类型系统。你将获得一个这样的编译器，它将阻止你引入许多与并发相关的错误，并在运行时为你提供出色的性能特征。开始使用 Pony如果你准备好开始使用 Pony，你需要先在 Pony 的网站上访问 学习部分。在这里你会找到安装 Pony 编译器的步骤和学习这门语言的资源。如果你愿意为你正在使用的这个语言做出贡献，我们会在 GitHub 上为你提供一些 初学者友好的问题。同时，我迫不及待地想在 我们的 IRC 频道 和 Pony 邮件列表 上与你交谈。要了解更多有关 Pony 的消息，请参阅 Sean Allen 2018 年 7 月 16 日至 19 日在俄勒冈州波特兰举行的 第 20 届 OSCON 会议 上的演讲： Pony，我如何学会停止担心并拥抱未经证实的技术。via: https://opensource.com/article/18/5/pony作者：Sean T Allen 选题：lujun9972 译者：beamrolling 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
Go 是一个内置支持并发编程的语言。借助使用 go 关键字去创建 协程 goroutine （轻量级线程）和在 Go 中提供的 使用 信道 和 其它的并发 同步方法，使得并发编程变得很容易、很灵活和很有趣。另一方面，Go 并不会阻止一些因 Go 程序员粗心大意或者缺乏经验而造成的并发编程错误。在本文的下面部分将展示一些在 Go 编程中常见的并发编程错误，以帮助 Go 程序员们避免再犯类似的错误。需要同步的时候没有同步代码行或许 不是按出现的顺序运行的。在下面的程序中有两个错误。第一，在 main 协程中读取 b 和在新的 协程 中写入 b 可能导致数据争用。第二，条件 b == true 并不能保证在 main 协程 中的 a != nil。在新的协程中编译器和 CPU 可能会通过 重排序指令 进行优化，因此，在运行时 b 赋值可能发生在 a 赋值之前，在 main 协程 中当 a 被修改后，它将会让部分 a 一直保持为 nil。package main import ( "time" "runtime" ) func main() { var a []int // nil var b bool // false // a new goroutine go func () { a = make([]int, 3) b = true // write b }() for !b { // read b time.Sleep(time.Second) runtime.Gosched() } a[0], a[1], a[2] = 0, 1, 2 // might panic } 上面的程序或者在一台计算机上运行的很好，但是在另一台上可能会引发异常。或者它可能运行了 N 次都很好，但是可能在第 (N+1) 次引发了异常。我们将使用 sync 标准包中提供的信道或者同步方法去确保内存中的顺序。例如，package main func main() { var a []int = nil c := make(chan struct{}) // a new goroutine go func () { a = make([]int, 3) c <- struct{}{} }() <-c a[0], a[1], a[2] = 0, 1, 2 } 使用 time.Sleep 调用去做同步我们先来看一个简单的例子。package main import ( "fmt" "time" ) func main() { var x = 123 go func() { x = 789 // write x }() time.Sleep(time.Second) fmt.Println(x) // read x } 我们预期程序将打印出 789。如果我们运行它，通常情况下，它确定打印的是 789。但是，这个程序使用的同步方式好吗？No！原因是 Go 运行时并不保证 x 的写入一定会发生在 x 的读取之前。在某些条件下，比如在同一个操作系统上，大部分 CPU 资源被其它运行的程序所占用的情况下，写入 x 可能就会发生在读取 x 之后。这就是为什么我们在正式的项目中，从来不使用 time.Sleep 调用去实现同步的原因。我们来看一下另外一个示例。package main import ( "fmt" "time" ) var x = 0 func main() { var num = 123 var p = &num c := make(chan int) go func() { c <- *p + x }() time.Sleep(time.Second) num = 789 fmt.Println(<-c) } 你认为程序的预期输出是什么？123 还是 789？事实上它的输出与编译器有关。对于标准的 Go 编译器 1.10 来说，这个程序很有可能输出是 123。但是在理论上，它可能输出的是 789，或者其它的随机数。现在，我们来改变 c <- *p + x 为 c <- *p，然后再次运行这个程序。你将会发现输出变成了 789 （使用标准的 Go 编译器 1.10）。这再次说明它的输出是与编译器相关的。是的，在上面的程序中存在数据争用。表达式 *p 可能会被先计算、后计算、或者在处理赋值语句 num = 789 时计算。time.Sleep 调用并不能保证 *p 发生在赋值语句处理之前进行。对于这个特定的示例，我们将在新的协程创建之前，将值保存到一个临时值中，然后在新的协程中使用临时值去消除数据争用。... tmp := *p + x go func() { c <- tmp }() ... 使协程挂起挂起协程是指让协程一直处于阻塞状态。导致协程被挂起的原因很多。比如，一个协程尝试从一个 nil 信道中或者从一个没有其它协程给它发送值的信道中检索数据。一个协程尝试去发送一个值到 nil 信道，或者发送到一个没有其它的协程接收值的信道中。一个协程被它自己死锁。一组协程彼此死锁。当运行一个没有 default 分支的 select 代码块时，一个协程被阻塞，以及在 select 代码块中 case 关键字后的所有信道操作保持阻塞状态。除了有时我们为了避免程序退出，特意让一个程序中的 main 协程保持挂起之外，大多数其它的协程挂起都是意外情况。Go 运行时很难判断一个协程到底是处于挂起状态还是临时阻塞。因此，Go 运行时并不会去释放一个挂起的协程所占用的资源。在 谁先响应谁获胜 的信道使用案例中，如果使用的 future 信道容量不够大，当尝试向 Future 信道发送结果时，一些响应较慢的信道将被挂起。比如，如果调用下面的函数，将有 4 个协程处于永远阻塞状态。func request() int { c := make(chan int) for i := 0; i < 5; i++ { i := i go func() { c <- i // 4 goroutines will hang here. }() } return <-c } 为避免这 4 个协程一直处于挂起状态， c 信道的容量必须至少是 4。在 实现谁先响应谁获胜的第二种方法 的信道使用案例中，如果将 future 信道用做非缓冲信道，那么有可能这个信息将永远也不会有响应而挂起。例如，如果在一个协程中调用下面的函数，协程可能会挂起。原因是，如果接收操作 <-c 准备就绪之前，五个发送操作全部尝试发送，那么所有的尝试发送的操作将全部失败，因此那个调用者协程将永远也不会接收到值。func request() int { c := make(chan int) for i := 0; i < 5; i++ { i := i go func() { select { case c <- i: default: } }() } return <-c } 将信道 c 变成缓冲信道将保证五个发送操作中的至少一个操作会发送成功，这样，上面函数中的那个调用者协程将不会被挂起。在 sync 标准包中拷贝类型值在实践中，sync 标准包中的类型值不会被拷贝。我们应该只拷贝这个值的指针。下面是一个错误的并发编程示例。在这个示例中，当调用 Counter.Value 方法时，将拷贝一个 Counter 接收值。作为接收值的一个字段，Counter 接收值的各个 Mutex 字段也会被拷贝。拷贝不是同步发生的，因此，拷贝的 Mutex 值可能会出错。即便是没有错误，拷贝的 Counter 接收值的访问保护也是没有意义的。import "sync" type Counter struct { sync.Mutex n int64 } // This method is okay. func (c *Counter) Increase(d int64) (r int64) { c.Lock() c.n += d r = c.n c.Unlock() return } // The method is bad. When it is called, a Counter // receiver value will be copied. func (c Counter) Value() (r int64) { c.Lock() r = c.n c.Unlock() return } 我们只需要改变 Value 接收类型方法为指针类型 *Counter，就可以避免拷贝 Mutex 值。在官方的 Go SDK 中提供的 go vet 命令将会报告潜在的错误值拷贝。在错误的地方调用 sync.WaitGroup 的方法每个 sync.WaitGroup 值维护一个内部计数器，这个计数器的初始值为 0。如果一个 WaitGroup 计数器的值是 0，调用 WaitGroup 值的 Wait 方法就不会被阻塞，否则，在计数器值为 0 之前，这个调用会一直被阻塞。为了让 WaitGroup 值的使用有意义，当一个 WaitGroup 计数器值为 0 时，必须在相应的 WaitGroup 值的 Wait 方法调用之前，去调用 WaitGroup 值的 Add 方法。例如，下面的程序中，在不正确位置调用了 Add 方法，这将使最后打印出的数字不总是 100。事实上，这个程序最后打印的数字可能是在 [0, 100) 范围内的一个随意数字。原因就是 Add 方法的调用并不保证一定会发生在 Wait 方法调用之前。package main import ( "fmt" "sync" "sync/atomic" ) func main() { var wg sync.WaitGroup var x int32 = 0 for i := 0; i < 100; i++ { go func() { wg.Add(1) atomic.AddInt32(&x, 1) wg.Done() }() } fmt.Println("To wait ...") wg.Wait() fmt.Println(atomic.LoadInt32(&x)) } 为让程序的表现符合预期，在 for 循环中，我们将把 Add 方法的调用移动到创建的新协程的范围之外，修改后的代码如下。... for i := 0; i < 100; i++ { wg.Add(1) go func() { atomic.AddInt32(&x, 1) wg.Done() }() } ... 不正确使用 futures 信道在 信道使用案例 的文章中，我们知道一些函数将返回 futures 信道。假设 fa 和 fb 就是这样的两个函数，那么下面的调用就使用了不正确的 future 参数。doSomethingWithFutureArguments(<-fa(), <-fb()) 在上面的代码行中，两个信道接收操作是顺序进行的，而不是并发的。我们做如下修改使它变成并发操作。ca, cb := fa(), fb() doSomethingWithFutureArguments(<-c1, <-c2) 没有等协程的最后的活动的发送结束就关闭信道Go 程序员经常犯的一个错误是，还有一些其它的协程可能会发送值到以前的信道时，这个信道就已经被关闭了。当这样的发送（发送到一个已经关闭的信道）真实发生时，将引发一个异常。这种错误在一些以往的著名 Go 项目中也有发生，比如在 Kubernetes 项目中的 这个 bug 和 这个 bug。如何安全和优雅地关闭信道，请阅读 这篇文章。在值上做 64 位原子操作时没有保证值地址 64 位对齐到目前为止（Go 1.10），在标准的 Go 编译器中，在一个 64 位原子操作中涉及到的值的地址要求必须是 64 位对齐的。如果没有对齐则导致当前的协程异常。对于标准的 Go 编译器来说，这种失败仅发生在 32 位的架构上。请阅读 内存布局 去了解如何在一个 32 位操作系统上保证 64 位对齐。没有注意到大量的资源被 time.After 函数调用占用在 time 标准包中的 After 函数返回 一个延迟通知的信道。这个函数在某些情况下用起来很便捷，但是，每次调用它将创建一个 time.Timer 类型的新值。这个新创建的 Timer 值在通过传递参数到 After 函数指定期间保持激活状态，如果在这个期间过多的调用了该函数，可能会有太多的 Timer 值保持激活，这将占用大量的内存和计算资源。例如，如果调用了下列的 longRunning 函数，将在一分钟内产生大量的消息，然后在某些周期内将有大量的 Timer 值保持激活，即便是大量的这些 Timer 值已经没用了也是如此。import ( "fmt" "time" ) // The function will return if a message arrival interval // is larger than one minute. func longRunning(messages <-chan string) { for { select { case <-time.After(time.Minute): return case msg := <-messages: fmt.Println(msg) } } } 为避免在上述代码中创建过多的 Timer 值，我们将使用一个单一的 Timer 值去完成同样的任务。func longRunning(messages <-chan string) { timer := time.NewTimer(time.Minute) defer timer.Stop() for { select { case <-timer.C: return case msg := <-messages: fmt.Println(msg) if !timer.Stop() { <-timer.C } } // The above "if" block can also be put here. timer.Reset(time.Minute) } } 不正确地使用 time.Timer 值在最后，我们将展示一个符合语言使用习惯的 time.Timer 值的使用示例。需要注意的一个细节是，那个 Reset 方法总是在停止或者 time.Timer 值释放时被使用。在 select 块的第一个 case 分支的结束部分，time.Timer 值被释放，因此，我们不需要去停止它。但是必须在第二个分支中停止定时器。如果在第二个分支中 if 代码块缺失，它可能至少在 Reset 方法调用时，会（通过 Go 运行时）发送到 timer.C 信道，并且那个 longRunning 函数可能会早于预期返回，对于 Reset 方法来说，它可能仅仅是重置内部定时器为 0，它将不会清理（耗尽）那个发送到 timer.C 信道的值。例如，下面的程序很有可能在一秒内而不是十秒时退出。并且更重要的是，这个程序并不是 DRF 的（LCTT 译注：data race free，多线程程序的一种同步程度）。package main import ( "fmt" "time" ) func main() { start := time.Now() timer := time.NewTimer(time.Second/2) select { case <-timer.C: default: time.Sleep(time.Second) // go here } timer.Reset(time.Second * 10) <-timer.C fmt.Println(time.Since(start)) // 1.000188181s } 当 time.Timer 的值不再被其它任何一个东西使用时，它的值可能被停留在一种非停止状态，但是，建议在结束时停止它。在多个协程中如果不按建议使用 time.Timer 值并发，可能会有 bug 隐患。我们不应该依赖一个 Reset 方法调用的返回值。Reset 方法返回值的存在仅仅是为了兼容性目的。via: https://go101.org/article/concurrent-common-mistakes.html作者：<go101.org> 译者：qhwdw 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是我写的并发网络服务器系列文章的第五部分。在前四部分中我们讨论了并发服务器的结构，这篇文章我们将去研究一个在生产系统中大量使用的服务器的案例—— Redis。Redis 是一个非常有魅力的项目，我关注它很久了。它最让我着迷的一点就是它的 C 源代码非常清晰。它也是一个高性能、大并发的内存数据库服务器的非常好的例子，它是研究网络并发服务器的一个非常好的案例，因此，我们不能错过这个好机会。我们来看看前四部分讨论的概念在真实世界中的应用程序。本系列的所有文章有：第一节 - 简介第二节 - 线程第三节 - 事件驱动第四节 - libuv第五节 - Redis 案例研究事件处理库Redis 最初发布于 2009 年，它最牛逼的一件事情大概就是它的速度 —— 它能够处理大量的并发客户端连接。需要特别指出的是，它是用一个单线程来完成的，而且还不对保存在内存中的数据使用任何复杂的锁或者同步机制。Redis 之所以如此牛逼是因为，它在给定的系统上使用了其可用的最快的事件循环，并将它们封装成由它实现的事件循环库（在 Linux 上是 epoll，在 BSD 上是 kqueue，等等）。这个库的名字叫做 ae。ae 使得编写一个快速服务器变得很容易，只要在它内部没有阻塞即可，而 Redis 则保证 注1 了这一点。在这里，我们的兴趣点主要是它对文件事件的支持 —— 当文件描述符（如网络套接字）有一些有趣的未决事情时将调用注册的回调函数。与 libuv 类似，ae 支持多路事件循环（参阅本系列的第三节和第四节）和不应该感到意外的 aeCreateFileEvent 信号：int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData); 它在 fd 上使用一个给定的事件循环，为新的文件事件注册一个回调（proc）函数。当使用的是 epoll 时，它将调用 epoll_ctl 在文件描述符上添加一个事件（可能是 EPOLLIN、EPOLLOUT、也或许两者都有，取决于 mask 参数）。ae 的 aeProcessEvents 功能是 “运行事件循环和发送回调函数”，它在底层调用了 epoll_wait。处理客户端请求我们通过跟踪 Redis 服务器代码来看一下，ae 如何为客户端事件注册回调函数的。initServer 启动时，通过注册一个回调函数来读取正在监听的套接字上的事件，通过使用回调函数 acceptTcpHandler 来调用 aeCreateFileEvent。当新的连接可用时，这个回调函数被调用。它调用 accept 注2 ，接下来是 acceptCommonHandler，它转而去调用 createClient 以初始化新客户端连接所需要的数据结构。createClient 的工作是去监听来自客户端的入站数据。它将套接字设置为非阻塞模式（一个异步事件循环中的关键因素）并使用 aeCreateFileEvent 去注册另外一个文件事件回调函数以读取事件 —— readQueryFromClient。每当客户端发送数据，这个函数将被事件循环调用。readQueryFromClient 就让我们期望的那样 —— 解析客户端命令和动作，并通过查询和/或操作数据来回复。因为客户端套接字是非阻塞的，所以这个函数必须能够处理 EAGAIN，以及部分数据；从客户端中读取的数据是累积在客户端专用的缓冲区中，而完整的查询可能被分割在回调函数的多个调用当中。将数据发送回客户端在前面的内容中，我说到了 readQueryFromClient 结束了发送给客户端的回复。这在逻辑上是正确的，因为 readQueryFromClient 准备要发送回复，但它不真正去做实质的发送 —— 因为这里并不能保证客户端套接字已经准备好写入/发送数据。我们必须为此使用事件循环机制。Redis 是这样做的，它注册一个 beforeSleep 函数，每次事件循环即将进入休眠时，调用它去等待套接字变得可以读取/写入。beforeSleep 做的其中一件事情就是调用 handleClientsWithPendingWrites。它的作用是通过调用 writeToClient 去尝试立即发送所有可用的回复；如果一些套接字不可用时，那么当套接字可用时，它将注册一个事件循环去调用 sendReplyToClient。这可以被看作为一种优化 —— 如果套接字可用于立即发送数据（一般是 TCP 套接字），这时并不需要注册事件 ——直接发送数据。因为套接字是非阻塞的，它从不会去阻塞循环。为什么 Redis 要实现它自己的事件库？在 第四节 中我们讨论了使用 libuv 来构建一个异步并发服务器。需要注意的是，Redis 并没有使用 libuv，或者任何类似的事件库，而是它去实现自己的事件库 —— ae，用 ae 来封装 epoll、kqueue 和 select。事实上，Antirez（Redis 的创建者）恰好在 2011 年的一篇文章 中回答了这个问题。他的回答的要点是：ae 只有大约 770 行他理解的非常透彻的代码；而 libuv 代码量非常巨大，也没有提供 Redis 所需的额外功能。现在，ae 的代码大约增长到 1300 多行，比起 libuv 的 26000 行（这是在没有 Windows、测试、示例、文档的情况下的数据）来说那是小巫见大巫了。libuv 是一个非常综合的库，这使它更复杂，并且很难去适应其它项目的特殊需求；另一方面，ae 是专门为 Redis 设计的，与 Redis 共同演进，只包含 Redis 所需要的东西。这是我 前些年在一篇文章中 提到的软件项目依赖关系的另一个很好的示例：依赖的优势与在软件项目上花费的工作量成反比。在某种程度上，Antirez 在他的文章中也提到了这一点。他提到，提供大量附加价值（在我的文章中的“基础” 依赖）的依赖比像 libuv 这样的依赖更有意义（它的例子是 jemalloc 和 Lua），对于 Redis 特定需求，其功能的实现相当容易。Redis 中的多线程在 Redis 的绝大多数历史中，它都是一个不折不扣的单线程的东西。一些人觉得这太不可思议了，有这种想法完全可以理解。Redis 本质上是受网络束缚的 —— 只要数据库大小合理，对于任何给定的客户端请求，其大部分延时都是浪费在网络等待上，而不是在 Redis 的数据结构上。然而，现在事情已经不再那么简单了。Redis 现在有几个新功能都用到了线程：“惰性” 内存释放。在后台线程中使用 fsync 调用写一个 持久化日志。运行需要执行一个长周期运行的操作的用户定义模块。对于前两个特性，Redis 使用它自己的一个简单的 bio（它是 “Background I/O" 的首字母缩写）库。这个库是根据 Redis 的需要进行了硬编码，它不能用到其它的地方 —— 它运行预设数量的线程，每个 Redis 后台作业类型需要一个线程。而对于第三个特性，Redis 模块 可以定义新的 Redis 命令，并且遵循与普通 Redis 命令相同的标准，包括不阻塞主线程。如果在模块中自定义的一个 Redis 命令，希望去执行一个长周期运行的操作，这将创建一个线程在后台去运行它。在 Redis 源码树中的 src/modules/helloblock.c 提供了这样的一个示例。有了这些特性，Redis 使用线程将一个事件循环结合起来，在一般的案例中，Redis 具有了更快的速度和弹性，这有点类似于在本系统文章中 第四节 讨论的工作队列。注1： Redis 的一个核心部分是：它是一个 内存中 数据库；因此，查询从不会运行太长的时间。当然了，这将会带来各种各样的其它问题。在使用分区的情况下，服务器可能最终路由一个请求到另一个实例上；在这种情况下，将使用异步 I/O 来避免阻塞其它客户端。注2： 使用 anetAccept；anet 是 Redis 对 TCP 套接字代码的封装。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-5-redis-case-study/作者：Eli Bendersky 译者：qhwdw 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是并发网络服务器系列文章的第四部分。在这一部分中，我们将使用 libuv 再次重写我们的服务器，并且也会讨论关于使用一个线程池在回调中去处理耗时任务。最终，我们去看一下底层的 libuv，花一点时间去学习如何用异步 API 对文件系统阻塞操作进行封装。本系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动第四节 - libuv使用 libuv 抽象出事件驱动循环在 第三节 中，我们看到了基于 select 和 epoll 的服务器的相似之处，并且，我说过，在它们之间抽象出细微的差别是件很有吸引力的事。许多库已经做到了这些，所以在这一部分中我将去选一个并使用它。我选的这个库是 libuv，它最初设计用于 Node.js 底层的可移植平台层，并且，后来发现在其它的项目中也有使用。libuv 是用 C 写的，因此，它具有很高的可移植性，非常适用嵌入到像 JavaScript 和 Python 这样的高级语言中。虽然 libuv 为了抽象出底层平台细节已经变成了一个相当大的框架，但它仍然是以 事件循环 思想为中心的。在我们第三部分的事件驱动服务器中，事件循环是显式定义在 main 函数中的；当使用 libuv 时，该循环通常隐藏在库自身中，而用户代码仅需要注册事件句柄（作为一个回调函数）和运行这个循环。此外，libuv 会在给定的平台上使用更快的事件循环实现，对于 Linux 它是 epoll，等等。libuv 支持多路事件循环，因此事件循环在库中是非常重要的；它有一个句柄 —— uv_loop_t，以及创建/杀死/启动/停止循环的函数。也就是说，在这篇文章中，我将仅需要使用 “默认的” 循环，libuv 可通过 uv_default_loop() 提供它；多路循环大多用于多线程事件驱动的服务器，这是一个更高级别的话题，我将留在这一系列文章的以后部分。使用 libuv 的并发服务器为了对 libuv 有一个更深的印象，让我们跳转到我们的可靠协议的服务器，它通过我们的这个系列已经有了一个强大的重新实现。这个服务器的结构与第三部分中的基于 select 和 epoll 的服务器有一些相似之处，因为，它也依赖回调。完整的 示例代码在这里；我们开始设置这个服务器的套接字绑定到一个本地端口：int portnum = 9090; if (argc >= 2) { portnum = atoi(argv[1]); } printf("Serving on port %d\n", portnum); int rc; uv_tcp_t server_stream; if ((rc = uv_tcp_init(uv_default_loop(), &server_stream)) < 0) { die("uv_tcp_init failed: %s", uv_strerror(rc)); } struct sockaddr_in server_address; if ((rc = uv_ip4_addr("0.0.0.0", portnum, &server_address)) < 0) { die("uv_ip4_addr failed: %s", uv_strerror(rc)); } if ((rc = uv_tcp_bind(&server_stream, (const struct sockaddr*)&server_address, 0)) < 0) { die("uv_tcp_bind failed: %s", uv_strerror(rc)); } 除了它被封装进 libuv API 中之外，你看到的是一个相当标准的套接字。在它的返回中，我们取得了一个可工作于任何 libuv 支持的平台上的可移植接口。这些代码也展示了很认真负责的错误处理；多数的 libuv 函数返回一个整数状态，返回一个负数意味着出现了一个错误。在我们的服务器中，我们把这些错误看做致命问题进行处理，但也可以设想一个更优雅的错误恢复。现在，那个套接字已经绑定，是时候去监听它了。这里我们运行首个回调注册：// Listen on the socket for new peers to connect. When a new peer connects, // the on_peer_connected callback will be invoked. if ((rc = uv_listen((uv_stream_t*)&server_stream, N_BACKLOG, on_peer_connected)) < 0) { die("uv_listen failed: %s", uv_strerror(rc)); } uv_listen 注册一个事件回调，当新的对端连接到这个套接字时将会调用事件循环。我们的回调在这里被称为 on_peer_connected，我们一会儿将去查看它。最终，main 运行这个 libuv 循环，直到它被停止（uv_run 仅在循环被停止或者发生错误时返回）。// Run the libuv event loop. uv_run(uv_default_loop(), UV_RUN_DEFAULT); // If uv_run returned, close the default loop before exiting. return uv_loop_close(uv_default_loop()); 注意，在运行事件循环之前，只有一个回调是通过 main 注册的；我们稍后将看到怎么去添加更多的回调。在事件循环的整个运行过程中，添加和删除回调并不是一个问题 —— 事实上，大多数服务器就是这么写的。这是一个 on_peer_connected，它处理到服务器的新的客户端连接：void on_peer_connected(uv_stream_t* server_stream, int status) { if (status < 0) { fprintf(stderr, "Peer connection error: %s\n", uv_strerror(status)); return; } // client will represent this peer; it's allocated on the heap and only // released when the client disconnects. The client holds a pointer to // peer_state_t in its data field; this peer state tracks the protocol state // with this client throughout interaction. uv_tcp_t* client = (uv_tcp_t*)xmalloc(sizeof(*client)); int rc; if ((rc = uv_tcp_init(uv_default_loop(), client)) < 0) { die("uv_tcp_init failed: %s", uv_strerror(rc)); } client->data = NULL; if (uv_accept(server_stream, (uv_stream_t*)client) == 0) { struct sockaddr_storage peername; int namelen = sizeof(peername); if ((rc = uv_tcp_getpeername(client, (struct sockaddr*)&peername, &namelen)) < 0) { die("uv_tcp_getpeername failed: %s", uv_strerror(rc)); } report_peer_connected((const struct sockaddr_in*)&peername, namelen); // Initialize the peer state for a new client: we start by sending the peer // the initial '*' ack. peer_state_t* peerstate = (peer_state_t*)xmalloc(sizeof(*peerstate)); peerstate->state = INITIAL_ACK; peerstate->sendbuf[0] = '*'; peerstate->sendbuf_end = 1; peerstate->client = client; client->data = peerstate; // Enqueue the write request to send the ack; when it's done, // on_wrote_init_ack will be called. The peer state is passed to the write // request via the data pointer; the write request does not own this peer // state - it's owned by the client handle. uv_buf_t writebuf = uv_buf_init(peerstate->sendbuf, peerstate->sendbuf_end); uv_write_t* req = (uv_write_t*)xmalloc(sizeof(*req)); req->data = peerstate; if ((rc = uv_write(req, (uv_stream_t*)client, &writebuf, 1, on_wrote_init_ack)) < 0) { die("uv_write failed: %s", uv_strerror(rc)); } } else { uv_close((uv_handle_t*)client, on_client_closed); } } 这些代码都有很好的注释，但是，这里有一些重要的 libuv 语法我想去强调一下：传入自定义数据到回调中：因为 C 语言还没有闭包，这可能是个挑战，libuv 在它的所有的处理类型中有一个 void* data 字段；这些字段可以被用于传递用户数据。例如，注意 client->data 是如何指向到一个 peer_state_t 结构上，以便于 uv_write 和 uv_read_start 注册的回调可以知道它们正在处理的是哪个客户端的数据。内存管理：在带有垃圾回收的语言中进行事件驱动编程是非常容易的，因为，回调通常运行在一个与它们注册的地方完全不同的栈帧中，使得基于栈的内存管理很困难。它总是需要传递堆分配的数据到 libuv 回调中（当所有回调运行时，除了 main，其它的都运行在栈上），并且，为了避免泄漏，许多情况下都要求这些数据去安全释放（free()）。这些都是些需要实践的内容 注1 。这个服务器上对端的状态如下：typedef struct { ProcessingState state; char sendbuf[SENDBUF_SIZE]; int sendbuf_end; uv_tcp_t* client; } peer_state_t; 它与第三部分中的状态非常类似；我们不再需要 sendptr，因为，在调用 “done writing” 回调之前，uv_write 将确保发送它提供的整个缓冲。我们也为其它的回调使用保持了一个到客户端的指针。这里是 on_wrote_init_ack：void on_wrote_init_ack(uv_write_t* req, int status) { if (status) { die("Write error: %s\n", uv_strerror(status)); } peer_state_t* peerstate = (peer_state_t*)req->data; // Flip the peer state to WAIT_FOR_MSG, and start listening for incoming data // from this peer. peerstate->state = WAIT_FOR_MSG; peerstate->sendbuf_end = 0; int rc; if ((rc = uv_read_start((uv_stream_t*)peerstate->client, on_alloc_buffer, on_peer_read)) < 0) { die("uv_read_start failed: %s", uv_strerror(rc)); } // Note: the write request doesn't own the peer state, hence we only free the // request itself, not the state. free(req); } 然后，我们确信知道了这个初始的 '*' 已经被发送到对端，我们通过调用 uv_read_start 去监听从这个对端来的入站数据，它注册一个将被事件循环调用的回调（on_peer_read），不论什么时候，事件循环都在套接字上接收来自客户端的调用：void on_peer_read(uv_stream_t* client, ssize_t nread, const uv_buf_t* buf) { if (nread < 0) { if (nread != uv_eof) { fprintf(stderr, "read error: %s\n", uv_strerror(nread)); } uv_close((uv_handle_t*)client, on_client_closed); } else if (nread == 0) { // from the documentation of uv_read_cb: nread might be 0, which does not // indicate an error or eof. this is equivalent to eagain or ewouldblock // under read(2). } else { // nread > 0 assert(buf->len >= nread); peer_state_t* peerstate = (peer_state_t*)client->data; if (peerstate->state == initial_ack) { // if the initial ack hasn't been sent for some reason, ignore whatever // the client sends in. free(buf->base); return; } // run the protocol state machine. for (int i = 0; i < nread; ++i) { switch (peerstate->state) { case initial_ack: assert(0 && "can't reach here"); break; case wait_for_msg: if (buf->base[i] == '^') { peerstate->state = in_msg; } break; case in_msg: if (buf->base[i] == '$') { peerstate->state = wait_for_msg; } else { assert(peerstate->sendbuf_end < sendbuf_size); peerstate->sendbuf[peerstate->sendbuf_end++] = buf->base[i] + 1; } break; } } if (peerstate->sendbuf_end > 0) { // we have data to send. the write buffer will point to the buffer stored // in the peer state for this client. uv_buf_t writebuf = uv_buf_init(peerstate->sendbuf, peerstate->sendbuf_end); uv_write_t* writereq = (uv_write_t*)xmalloc(sizeof(*writereq)); writereq->data = peerstate; int rc; if ((rc = uv_write(writereq, (uv_stream_t*)client, &writebuf, 1, on_wrote_buf)) < 0) { die("uv_write failed: %s", uv_strerror(rc)); } } } free(buf->base); } 这个服务器的运行时行为非常类似于第三部分的事件驱动服务器：所有的客户端都在一个单个的线程中并发处理。并且类似的，一些特定的行为必须在服务器代码中维护：服务器的逻辑实现为一个集成的回调，并且长周期运行是禁止的，因为它会阻塞事件循环。这一点也很类似。让我们进一步探索这个问题。在事件驱动循环中的长周期运行的操作单线程的事件驱动代码使它先天就容易受到一些常见问题的影响：长周期运行的代码会阻塞整个循环。参见如下的程序：void on_timer(uv_timer_t* timer) { uint64_t timestamp = uv_hrtime(); printf("on_timer [%" PRIu64 " ms]\n", (timestamp / 1000000) % 100000); // "Work" if (random() % 5 == 0) { printf("Sleeping...\n"); sleep(3); } } int main(int argc, const char** argv) { uv_timer_t timer; uv_timer_init(uv_default_loop(), &timer); uv_timer_start(&timer, on_timer, 0, 1000); return uv_run(uv_default_loop(), UV_RUN_DEFAULT); } 它用一个单个注册的回调运行一个 libuv 事件循环：on_timer，它被每秒钟循环调用一次。回调报告一个时间戳，并且，偶尔通过睡眠 3 秒去模拟一个长周期运行。这是运行示例：$ ./uv-timer-sleep-demo on_timer [4840 ms] on_timer [5842 ms] on_timer [6843 ms] on_timer [7844 ms] Sleeping... on_timer [11845 ms] on_timer [12846 ms] Sleeping... on_timer [16847 ms] on_timer [17849 ms] on_timer [18850 ms] ... on_timer 忠实地每秒执行一次，直到随机出现的睡眠为止。在那个时间点，on_timer 不再被调用，直到睡眠时间结束；事实上，没有其它的回调 会在这个时间帧中被调用。这个睡眠调用阻塞了当前线程，它正是被调用的线程，并且也是事件循环使用的线程。当这个线程被阻塞后，事件循环也被阻塞。这个示例演示了在事件驱动的调用中为什么回调不能被阻塞是多少的重要。并且，同样适用于 Node.js 服务器、客户端侧的 Javascript、大多数的 GUI 编程框架、以及许多其它的异步编程模型。但是，有时候运行耗时的任务是不可避免的。并不是所有任务都有一个异步 API；例如，我们可能使用一些仅有同步 API 的库去处理，或者，正在执行一个可能的长周期计算。我们如何用事件驱动编程去结合这些代码？线程可以帮到你！“转换” 阻塞调用为异步调用的线程一个线程池可以用于转换阻塞调用为异步调用，通过与事件循环并行运行，并且当任务完成时去由它去公布事件。以阻塞函数 do_work() 为例，这里介绍了它是怎么运行的：不在一个回调中直接调用 do_work() ，而是将它打包进一个 “任务”，让线程池去运行这个任务。当任务完成时，我们也为循环去调用它注册一个回调；我们称它为 on_work_done()。在这个时间点，我们的回调就可以返回了，而事件循环保持运行；在同一时间点，线程池中的有一个线程运行这个任务。一旦任务运行完成，通知主线程（指正在运行事件循环的线程），并且事件循环调用 on_work_done()。让我们看一下，使用 libuv 的工作调度 API，是怎么去解决我们前面的计时器/睡眠示例中展示的问题的：void on_after_work(uv_work_t* req, int status) { free(req); } void on_work(uv_work_t* req) { // "Work" if (random() % 5 == 0) { printf("Sleeping...\n"); sleep(3); } } void on_timer(uv_timer_t* timer) { uint64_t timestamp = uv_hrtime(); printf("on_timer [%" PRIu64 " ms]\n", (timestamp / 1000000) % 100000); uv_work_t* work_req = (uv_work_t*)malloc(sizeof(*work_req)); uv_queue_work(uv_default_loop(), work_req, on_work, on_after_work); } int main(int argc, const char** argv) { uv_timer_t timer; uv_timer_init(uv_default_loop(), &timer); uv_timer_start(&timer, on_timer, 0, 1000); return uv_run(uv_default_loop(), UV_RUN_DEFAULT); } 通过一个 work_req 注2 类型的句柄，我们进入一个任务队列，代替在 on_timer 上直接调用 sleep，这个函数在任务中（on_work）运行，并且，一旦任务完成（on_after_work），这个函数被调用一次。on_work 是指 “work”（阻塞中的/耗时的操作）进行的地方。注意在这两个回调传递到 uv_queue_work 时的一个关键区别：on_work 运行在线程池中，而 on_after_work 运行在事件循环中的主线程上 —— 就好像是其它的回调一样。让我们看一下这种方式的运行：$ ./uv-timer-work-demo on_timer [89571 ms] on_timer [90572 ms] on_timer [91573 ms] on_timer [92575 ms] Sleeping... on_timer [93576 ms] on_timer [94577 ms] Sleeping... on_timer [95577 ms] on_timer [96578 ms] on_timer [97578 ms] ... 即便在 sleep 函数被调用时，定时器也每秒钟滴答一下，睡眠现在运行在一个单独的线程中，并且不会阻塞事件循环。一个用于练习的素数测试服务器因为通过睡眠去模拟工作并不是件让人兴奋的事，我有一个事先准备好的更综合的一个示例 —— 一个基于套接字接受来自客户端的数字的服务器，检查这个数字是否是素数，然后去返回一个 “prime" 或者 “composite”。完整的 服务器代码在这里 —— 我不在这里粘贴了，因为它太长了，更希望读者在一些自己的练习中去体会它。这个服务器使用了一个原生的素数测试算法，因此，对于大的素数可能花很长时间才返回一个回答。在我的机器中，对于 2305843009213693951，它花了 ~5 秒钟去计算，但是，你的方法可能不同。练习 1：服务器有一个设置（通过一个名为 MODE 的环境变量）要么在套接字回调（意味着在主线程上）中运行素数测试，要么在 libuv 工作队列中。当多个客户端同时连接时，使用这个设置来观察服务器的行为。当它计算一个大的任务时，在阻塞模式中，服务器将不回复其它客户端，而在非阻塞模式中，它会回复。练习 2：libuv 有一个缺省大小的线程池，并且线程池的大小可以通过环境变量配置。你可以通过使用多个客户端去实验找出它的缺省值是多少？找到线程池缺省值后，使用不同的设置去看一下，在重负载下怎么去影响服务器的响应能力。在非阻塞文件系统中使用工作队列对于只是呆板的演示和 CPU 密集型的计算来说，将可能的阻塞操作委托给一个线程池并不是明智的；libuv 在它的文件系统 API 中本身就大量使用了这种能力。通过这种方式，libuv 使用一个异步 API，以一个轻便的方式显示出它强大的文件系统的处理能力。让我们使用 uv_fs_read()，例如，这个函数从一个文件中（表示为一个 uv_fs_t 句柄）读取一个文件到一个缓冲中 注3，并且当读取完成后调用一个回调。换句话说， uv_fs_read() 总是立即返回，即使是文件在一个类似 NFS 的系统上，而数据到达缓冲区可能需要一些时间。换句话说，这个 API 与这种方式中其它的 libuv API 是异步的。这是怎么工作的呢？在这一点上，我们看一下 libuv 的底层；内部实际上非常简单，并且它是一个很好的练习。作为一个可移植的库，libuv 对于 Windows 和 Unix 系统在它的许多函数上有不同的实现。我们去看一下在 libuv 源树中的 src/unix/fs.c。这是 uv_fs_read 的代码：int uv_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_file file, const uv_buf_t bufs[], unsigned int nbufs, int64_t off, uv_fs_cb cb) { if (bufs == NULL || nbufs == 0) return -EINVAL; INIT(READ); req->file = file; req->nbufs = nbufs; req->bufs = req->bufsml; if (nbufs > ARRAY_SIZE(req->bufsml)) req->bufs = uv__malloc(nbufs * sizeof(*bufs)); if (req->bufs == NULL) { if (cb != NULL) uv__req_unregister(loop, req); return -ENOMEM; } memcpy(req->bufs, bufs, nbufs * sizeof(*bufs)); req->off = off; POST; } 第一次看可能觉得很困难，因为它延缓真实的工作到 INIT 和 POST 宏中，以及为 POST 设置了一些本地变量。这样做可以避免了文件中的许多重复代码。这是 INIT 宏：#define INIT(subtype) \ do { \ req->type = UV_FS; \ if (cb != NULL) \ uv__req_init(loop, req, UV_FS); \ req->fs_type = UV_FS_ ## subtype; \ req->result = 0; \ req->ptr = NULL; \ req->loop = loop; \ req->path = NULL; \ req->new_path = NULL; \ req->cb = cb; \ } \ while (0) 它设置了请求，并且更重要的是，设置 req->fs_type 域为真实的 FS 请求类型。因为 uv_fs_read 调用 INIT(READ)，它意味着 req->fs_type 被分配一个常数 UV_FS_READ。这是 POST 宏：#define POST \ do { \ if (cb != NULL) { \ uv__work_submit(loop, &req->work_req, uv__fs_work, uv__fs_done); \ return 0; \ } \ else { \ uv__fs_work(&req->work_req); \ return req->result; \ } \ } \ while (0) 它做什么取决于回调是否为 NULL。在 libuv 文件系统 API 中，一个 NULL 回调意味着我们真实地希望去执行一个 同步 操作。在这种情况下，POST 直接调用 uv__fs_work（我们需要了解一下这个函数的功能），而对于一个非 NULL 回调，它把 uv__fs_work 作为一个工作项提交到工作队列（指的是线程池），然后，注册 uv__fs_done 作为回调；该函数执行一些登记并调用用户提供的回调。如果我们去看 uv__fs_work 的代码，我们将看到它使用很多宏按照需求将工作分发到实际的文件系统调用。在我们的案例中，对于 UV_FS_READ 这个调用将被 uv__fs_read 生成，它（最终）使用普通的 POSIX API 去读取。这个函数可以在一个 阻塞 方式中很安全地实现。因为，它通过异步 API 调用时被置于一个线程池中。在 Node.js 中，fs.readFile 函数是映射到 uv_fs_read 上。因此，可以在一个非阻塞模式中读取文件，甚至是当底层文件系统 API 是阻塞方式时。注1： 为确保服务器不泄露内存，我在一个启用泄露检查的 Valgrind 中运行它。因为服务器经常是被设计为永久运行，这是一个挑战；为克服这个问题，我在服务器上添加了一个 “kill 开关” —— 一个从客户端接收的特定序列，以使它可以停止事件循环并退出。这个代码在 theon_wrote_buf 句柄中。注2： 在这里我们不过多地使用 work_req；讨论的素数测试服务器接下来将展示怎么被用于去传递上下文信息到回调中。注3： uv_fs_read() 提供了一个类似于 preadv Linux 系统调用的通用 API：它使用多缓冲区用于排序，并且支持一个到文件中的偏移。基于我们讨论的目的可以忽略这些特性。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-4-libuv/作者：Eli Bendersky 译者：qhwdw 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是并发服务器系列的第三节。第一节 介绍了阻塞式编程，第二节：线程 探讨了多线程，将其作为一种可行的方法来实现服务器并发编程。另一种常见的实现并发的方法叫做 事件驱动编程，也可以叫做 异步 编程 注1 。这种方法变化万千，因此我们会从最基本的开始，使用一些基本的 API 而非从封装好的高级方法开始。本系列以后的文章会讲高层次抽象，还有各种混合的方法。本系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动阻塞式 vs. 非阻塞式 I/O作为本篇的介绍，我们先讲讲阻塞和非阻塞 I/O 的区别。阻塞式 I/O 更好理解，因为这是我们使用 I/O 相关 API 时的“标准”方式。从套接字接收数据的时候，调用 recv 函数会发生 阻塞，直到它从端口上接收到了来自另一端套接字的数据。这恰恰是第一部分讲到的顺序服务器的问题。因此阻塞式 I/O 存在着固有的性能问题。第二节里我们讲过一种解决方法，就是用多线程。哪怕一个线程的 I/O 阻塞了，别的线程仍然可以使用 CPU 资源。实际上，阻塞 I/O 通常在利用资源方面非常高效，因为线程就等待着 —— 操作系统将线程变成休眠状态，只有满足了线程需要的条件才会被唤醒。非阻塞式 I/O 是另一种思路。把套接字设成非阻塞模式时，调用 recv 时（还有 send，但是我们现在只考虑接收），函数返回的会很快，哪怕没有接收到数据。这时，就会返回一个特殊的错误状态 注2 来通知调用者，此时没有数据传进来。调用者可以去做其他的事情，或者尝试再次调用 recv 函数。示范阻塞式和非阻塞式的 recv 区别的最好方式就是贴一段示例代码。这里有个监听套接字的小程序，一直在 recv 这里阻塞着；当 recv 返回了数据，程序就报告接收到了多少个字节 注3 ：int main(int argc, const char** argv) { setvbuf(stdout, NULL, _IONBF, 0); int portnum = 9988; if (argc >= 2) { portnum = atoi(argv[1]); } printf("Listening on port %d\n", portnum); int sockfd = listen_inet_socket(portnum); struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); while (1) { uint8_t buf[1024]; printf("Calling recv...\n"); int len = recv(newsockfd, buf, sizeof buf, 0); if (len < 0) { perror_die("recv"); } else if (len == 0) { printf("Peer disconnected; I'm done.\n"); break; } printf("recv returned %d bytes\n", len); } close(newsockfd); close(sockfd); return 0; } 主循环重复调用 recv 并且报告它返回的字节数（记住 recv 返回 0 时，就是客户端断开连接了）。试着运行它，我们会在一个终端里运行这个程序，然后在另一个终端里用 nc 进行连接，发送一些字符，每次发送之间间隔几秒钟：$ nc localhost 9988 hello # wait for 2 seconds after typing this socket world ^D # to end the connection> 监听程序会输出以下内容：$ ./blocking-listener 9988 Listening on port 9988 peer (localhost, 37284) connected Calling recv... recv returned 6 bytes Calling recv... recv returned 13 bytes Calling recv... Peer disconnected; I'm done. 现在试试非阻塞的监听程序的版本。这是代码：int main(int argc, const char** argv) { setvbuf(stdout, NULL, _IONBF, 0); int portnum = 9988; if (argc >= 2) { portnum = atoi(argv[1]); } printf("Listening on port %d\n", portnum); int sockfd = listen_inet_socket(portnum); struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); // 把套接字设成非阻塞模式 int flags = fcntl(newsockfd, F_GETFL, 0); if (flags == -1) { perror_die("fcntl F_GETFL"); } if (fcntl(newsockfd, F_SETFL, flags | O_NONBLOCK) == -1) { perror_die("fcntl F_SETFL O_NONBLOCK"); } while (1) { uint8_t buf[1024]; printf("Calling recv...\n"); int len = recv(newsockfd, buf, sizeof buf, 0); if (len < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { usleep(200 * 1000); continue; } perror_die("recv"); } else if (len == 0) { printf("Peer disconnected; I'm done.\n"); break; } printf("recv returned %d bytes\n", len); } close(newsockfd); close(sockfd); return 0; } 这里与阻塞版本有些差异，值得注意：accept 函数返回的 newsocktfd 套接字因调用了 fcntl， 被设置成非阻塞的模式。检查 recv 的返回状态时，我们对 errno 进行了检查，判断它是否被设置成表示没有可供接收的数据的状态。这时，我们仅仅是休眠了 200 毫秒然后进入到下一轮循环。同样用 nc 进行测试，以下是非阻塞监听器的输出：$ ./nonblocking-listener 9988 Listening on port 9988 peer (localhost, 37288) connected Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... recv returned 6 bytes Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... recv returned 13 bytes Calling recv... Calling recv... Calling recv... Peer disconnected; I'm done. 作为练习，给输出添加一个时间戳，确认调用 recv 得到结果之间花费的时间是比输入到 nc 中所用的多还是少（每一轮是 200 ms）。这里就实现了使用非阻塞的 recv 让监听者检查套接字变为可能，并且在没有数据的时候重新获得控制权。换句话说，用编程的语言说这就是 轮询 polling —— 主程序周期性的查询套接字以便读取数据。对于顺序响应的问题，这似乎是个可行的方法。非阻塞的 recv 让同时与多个套接字通信变成可能，轮询这些套接字，仅当有新数据到来时才处理。就是这样，这种方式 可以 用来写并发服务器；但实际上一般不这么做，因为轮询的方式很难扩展。首先，我在代码中引入的 200ms 延迟对于演示非常好（监听器在我输入 nc 之间只打印几行 “Calling recv...”，但实际上应该有上千行）。但它也增加了多达 200ms 的服务器响应时间，这无意是不必要的。实际的程序中，延迟会低得多，休眠时间越短，进程占用的 CPU 资源就越多。有些时钟周期只是浪费在等待，这并不好，尤其是在移动设备上，这些设备的电量往往有限。但是当我们实际这样来使用多个套接字的时候，更严重的问题出现了。想像下监听器正在同时处理 1000 个客户端。这意味着每一个循环迭代里面，它都得为 这 1000 个套接字中的每一个 执行一遍非阻塞的 recv，找到其中准备好了数据的那一个。这非常低效，并且极大的限制了服务器能够并发处理的客户端数。这里有个准则：每次轮询之间等待的间隔越久，服务器响应性越差；而等待的时间越少，CPU 在无用的轮询上耗费的资源越多。讲真，所有的轮询都像是无用功。当然操作系统应该是知道哪个套接字是准备好了数据的，因此没必要逐个扫描。事实上，就是这样，接下来就会讲一些 API，让我们可以更优雅地处理多个客户端。selectselect 的系统调用是可移植的（POSIX），是标准 Unix API 中常有的部分。它是为上一节最后一部分描述的问题而设计的 —— 允许一个线程可以监视许多文件描述符 注4 的变化，而不用在轮询中执行不必要的代码。我并不打算在这里引入一个关于 select 的全面教程，有很多网站和书籍讲这个，但是在涉及到问题的相关内容时，我会介绍一下它的 API，然后再展示一个非常复杂的例子。select 允许 多路 I/O，监视多个文件描述符，查看其中任何一个的 I/O 是否可用。int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout); readfds 指向文件描述符的缓冲区，这个缓冲区被监视是否有读取事件；fd_set 是一个特殊的数据结构，用户使用 FD_* 宏进行操作。writefds 是针对写事件的。nfds 是监视的缓冲中最大的文件描述符数字（文件描述符就是整数）。timeout 可以让用户指定 select 应该阻塞多久，直到某个文件描述符准备好了（timeout == NULL 就是说一直阻塞着）。现在先跳过 exceptfds。select 的调用过程如下：在调用之前，用户先要为所有不同种类的要监视的文件描述符创建 fd_set 实例。如果想要同时监视读取和写入事件，readfds 和 writefds 都要被创建并且引用。用户可以使用 FD_SET 来设置集合中想要监视的特殊描述符。例如，如果想要监视描述符 2、7 和 10 的读取事件，在 readfds 这里调用三次 FD_SET，分别设置 2、7 和 10。select 被调用。当 select 返回时（现在先不管超时），就是说集合中有多少个文件描述符已经就绪了。它也修改 readfds 和 writefds 集合，来标记这些准备好的描述符。其它所有的描述符都会被清空。这时用户需要遍历 readfds 和 writefds，找到哪个描述符就绪了（使用 FD_ISSET）。作为完整的例子，我在并发的服务器程序上使用 select，重新实现了我们之前的协议。完整的代码在这里；接下来的是代码中的重点部分及注释。警告：示例代码非常复杂，因此第一次看的时候，如果没有足够的时间，快速浏览也没有关系。使用 select 的并发服务器使用 I/O 的多发 API 诸如 select 会给我们服务器的设计带来一些限制；这不会马上显现出来，但这值得探讨，因为它们是理解事件驱动编程到底是什么的关键。最重要的是，要记住这种方法本质上是单线程的 注5 。服务器实际上在 同一时刻只能做一件事。因为我们想要同时处理多个客户端请求，我们需要换一种方式重构代码。首先，让我们谈谈主循环。它看起来是什么样的呢？先让我们想象一下服务器有一堆任务，它应该监视哪些东西呢？两种类型的套接字活动：新客户端尝试连接。这些客户端应该被 accept。已连接的客户端发送数据。这个数据要用 第一节 中所讲到的协议进行传输，有可能会有一些数据要被回送给客户端。尽管这两种活动在本质上有所区别，我们还是要把它们放在一个循环里，因为只能有一个主循环。循环会包含 select 的调用。这个 select 的调用会监视上述的两种活动。这里是部分代码，设置了文件描述符集合，并在主循环里转到被调用的 select 部分。// “master” 集合存活在该循环中，跟踪我们想要监视的读取事件或写入事件的文件描述符（FD）。 fd_set readfds_master; FD_ZERO(&readfds_master); fd_set writefds_master; FD_ZERO(&writefds_master); // 监听的套接字一直被监视，用于读取数据，并监测到来的新的端点连接。 FD_SET(listener_sockfd, &readfds_master); // 要想更加高效，fdset_max 追踪当前已知最大的 FD；这使得每次调用时对 FD_SETSIZE 的迭代选择不是那么重要了。 int fdset_max = listener_sockfd; while (1) { // select() 会修改传递给它的 fd_sets，因此进行拷贝一下再传值。 fd_set readfds = readfds_master; fd_set writefds = writefds_master; int nready = select(fdset_max + 1, &readfds, &writefds, NULL, NULL); if (nready < 0) { perror_die("select"); } ... 这里的一些要点：由于每次调用 select 都会重写传递给函数的集合，调用器就得维护一个 “master” 集合，在循环迭代中，保持对所监视的所有活跃的套接字的追踪。注意我们所关心的，最开始的唯一那个套接字是怎么变成 listener_sockfd 的，这就是最开始的套接字，服务器借此来接收新客户端的连接。select 的返回值，是在作为参数传递的集合中，那些已经就绪的描述符的个数。select 修改这个集合，用来标记就绪的描述符。下一步是在这些描述符中进行迭代。... for (int fd = 0; fd <= fdset_max && nready > 0; fd++) { // 检查 fd 是否变成可读的 if (FD_ISSET(fd, &readfds)) { nready--; if (fd == listener_sockfd) { // 监听的套接字就绪了；这意味着有个新的客户端连接正在联系 ... } else { fd_status_t status = on_peer_ready_recv(fd); if (status.want_read) { FD_SET(fd, &readfds_master); } else { FD_CLR(fd, &readfds_master); } if (status.want_write) { FD_SET(fd, &writefds_master); } else { FD_CLR(fd, &writefds_master); } if (!status.want_read && !status.want_write) { printf("socket %d closing\n", fd); close(fd); } } 这部分循环检查 可读的 描述符。让我们跳过监听器套接字（要浏览所有内容，看这个代码） 然后看看当其中一个客户端准备好了之后会发生什么。出现了这种情况后，我们调用一个叫做 on_peer_ready_recv 的 回调 函数，传入相应的文件描述符。这个调用意味着客户端连接到套接字上，发送某些数据，并且对套接字上 recv 的调用不会被阻塞 注6 。这个回调函数返回结构体 fd_status_t。typedef struct { bool want_read; bool want_write; } fd_status_t; 这个结构体告诉主循环，是否应该监视套接字的读取事件、写入事件，或者两者都监视。上述代码展示了 FD_SET 和 FD_CLR 是怎么在合适的描述符集合中被调用的。对于主循环中某个准备好了写入数据的描述符，代码是类似的，除了它所调用的回调函数，这个回调函数叫做 on_peer_ready_send。现在来花点时间看看这个回调：typedef enum { INITIAL_ACK, WAIT_FOR_MSG, IN_MSG } ProcessingState; #define SENDBUF_SIZE 1024 typedef struct { ProcessingState state; // sendbuf 包含了服务器要返回给客户端的数据。on_peer_ready_recv 句柄填充这个缓冲， // on_peer_read_send 进行消耗。sendbuf_end 指向缓冲区的最后一个有效字节， // sendptr 指向下个字节 uint8_t sendbuf[SENDBUF_SIZE]; int sendbuf_end; int sendptr; } peer_state_t; // 每一端都是通过它连接的文件描述符（fd）进行区分。只要客户端连接上了，fd 就是唯一的。 // 当客户端断开连接，另一个客户端连接上就会获得相同的 fd。on_peer_connected 应该 // 进行初始化，以便移除旧客户端在同一个 fd 上留下的东西。 peer_state_t global_state[MAXFDS]; fd_status_t on_peer_ready_recv(int sockfd) { assert(sockfd < MAXFDs); peer_state_t* peerstate = &global_state[sockfd]; if (peerstate->state == INITIAL_ACK || peerstate->sendptr < peerstate->sendbuf_end) { // 在初始的 ACK 被送到了客户端，就没有什么要接收的了。 // 等所有待发送的数据都被发送之后接收更多的数据。 return fd_status_W; } uint8_t buf[1024]; int nbytes = recv(sockfd, buf, sizeof buf, 0); if (nbytes == 0) { // 客户端断开连接 return fd_status_NORW; } else if (nbytes < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { // 套接字 *实际* 并没有准备好接收，等到它就绪。 return fd_status_R; } else { perror_die("recv"); } } bool ready_to_send = false; for (int i = 0; i < nbytes; ++i) { switch (peerstate->state) { case INITIAL_ACK: assert(0 && "can't reach here"); break; case WAIT_FOR_MSG: if (buf[i] == '^') { peerstate->state = IN_MSG; } break; case IN_MSG: if (buf[i] == '$') { peerstate->state = WAIT_FOR_MSG; } else { assert(peerstate->sendbuf_end < SENDBUF_SIZE); peerstate->sendbuf[peerstate->sendbuf_end++] = buf[i] + 1; ready_to_send = true; } break; } } // 如果没有数据要发送给客户端，报告读取状态作为最后接收的结果。 return (fd_status_t){.want_read = !ready_to_send, .want_write = ready_to_send}; } peer_state_t 是全状态对象，用来表示在主循环中两次回调函数调用之间的客户端的连接。因为回调函数在客户端发送的某些数据时被调用，不能假设它能够不停地与客户端通信，并且它得运行得很快，不能被阻塞。因为套接字被设置成非阻塞模式，recv 会快速的返回。除了调用 recv， 这个句柄做的是处理状态，没有其它的调用，从而不会发生阻塞。举个例子，你知道为什么这个代码需要一个额外的状态吗？这个系列中，我们的服务器目前只用到了两个状态，但是这个服务器程序需要三个状态。来看看 “套接字准备好发送” 的回调函数：fd_status_t on_peer_ready_send(int sockfd) { assert(sockfd < MAXFDs); peer_state_t* peerstate = &global_state[sockfd]; if (peerstate->sendptr >= peerstate->sendbuf_end) { // 没有要发送的。 return fd_status_RW; } int sendlen = peerstate->sendbuf_end - peerstate->sendptr; int nsent = send(sockfd, peerstate->sendbuf, sendlen, 0); if (nsent == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) { return fd_status_W; } else { perror_die("send"); } } if (nsent < sendlen) { peerstate->sendptr += nsent; return fd_status_W; } else { // 所有东西都成功发送；重置发送队列。 peerstate->sendptr = 0; peerstate->sendbuf_end = 0; // 如果我们现在是处于特殊的 INITIAL_ACK 状态，就转变到其他状态。 if (peerstate->state == INITIAL_ACK) { peerstate->state = WAIT_FOR_MSG; } return fd_status_R; } } 这里也一样，回调函数调用了一个非阻塞的 send，演示了状态管理。在异步代码中，回调函数执行的很快是受争议的，任何延迟都会阻塞主循环进行处理，因此也阻塞了整个服务器程序去处理其他的客户端。用脚步再来运行这个服务器，同时连接 3 个客户端。在一个终端中我们运行下面的命令：$ ./select-server 在另一个终端中：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-26 05:29:15,864:conn1 connected... INFO:2017-09-26 05:29:15,864:conn2 connected... INFO:2017-09-26 05:29:15,864:conn0 connected... INFO:2017-09-26 05:29:15,865:conn1 sending b'^abc$de^abte$f' INFO:2017-09-26 05:29:15,865:conn2 sending b'^abc$de^abte$f' INFO:2017-09-26 05:29:15,865:conn0 sending b'^abc$de^abte$f' INFO:2017-09-26 05:29:15,865:conn1 received b'bcdbcuf' INFO:2017-09-26 05:29:15,865:conn2 received b'bcdbcuf' INFO:2017-09-26 05:29:15,865:conn0 received b'bcdbcuf' INFO:2017-09-26 05:29:16,866:conn1 sending b'xyz^123' INFO:2017-09-26 05:29:16,867:conn0 sending b'xyz^123' INFO:2017-09-26 05:29:16,867:conn2 sending b'xyz^123' INFO:2017-09-26 05:29:16,867:conn1 received b'234' INFO:2017-09-26 05:29:16,868:conn0 received b'234' INFO:2017-09-26 05:29:16,868:conn2 received b'234' INFO:2017-09-26 05:29:17,868:conn1 sending b'25$^ab0000$abab' INFO:2017-09-26 05:29:17,869:conn1 received b'36bc1111' INFO:2017-09-26 05:29:17,869:conn0 sending b'25$^ab0000$abab' INFO:2017-09-26 05:29:17,870:conn0 received b'36bc1111' INFO:2017-09-26 05:29:17,870:conn2 sending b'25$^ab0000$abab' INFO:2017-09-26 05:29:17,870:conn2 received b'36bc1111' INFO:2017-09-26 05:29:18,069:conn1 disconnecting INFO:2017-09-26 05:29:18,070:conn0 disconnecting INFO:2017-09-26 05:29:18,070:conn2 disconnecting 和线程的情况相似，客户端之间没有延迟，它们被同时处理。而且在 select-server 也没有用线程！主循环 多路 处理所有的客户端，通过高效使用 select 轮询多个套接字。回想下 第二节中 顺序的 vs 多线程的客户端处理过程的图片。对于我们的 select-server，三个客户端的处理流程像这样：所有的客户端在同一个线程中同时被处理，通过乘积，做一点这个客户端的任务，然后切换到另一个，再切换到下一个，最后切换回到最开始的那个客户端。注意，这里没有什么循环调度，客户端在它们发送数据的时候被客户端处理，这实际上是受客户端左右的。同步、异步、事件驱动、回调select-server 示例代码为讨论什么是异步编程、它和事件驱动及基于回调的编程有何联系，提供了一个良好的背景。因为这些词汇在并发服务器的（非常矛盾的）讨论中很常见。让我们从一段 select 的手册页面中引用的一句话开始：select，pselect，FD\_CLR，FD\_ISSET，FD\_SET，FD\_ZERO - 同步 I/O 处理因此 select 是 同步 处理。但我刚刚演示了大量代码的例子，使用 select 作为 异步 处理服务器的例子。有哪些东西？答案是：这取决于你的观察角度。同步常用作阻塞处理，并且对 select 的调用实际上是阻塞的。和第 1、2 节中讲到的顺序的、多线程的服务器中对 send 和 recv 是一样的。因此说 select 是 同步的 API 是有道理的。可是，服务器的设计却可以是 异步的，或是 基于回调的，或是 事件驱动的，尽管其中有对 select 的使用。注意这里的 on_peer_* 函数是回调函数；它们永远不会阻塞，并且只有网络事件触发的时候才会被调用。它们可以获得部分数据，并能够在调用过程中保持稳定的状态。如果你曾经做过一些 GUI 编程，这些东西对你来说应该很亲切。有个 “事件循环”，常常完全隐藏在框架里，应用的 “业务逻辑” 建立在回调上，这些回调会在各种事件触发后被调用，用户点击鼠标、选择菜单、定时器触发、数据到达套接字等等。曾经最常见的编程模型是客户端的 JavaScript，这里面有一堆回调函数，它们在浏览网页时用户的行为被触发。select 的局限使用 select 作为第一个异步服务器的例子对于说明这个概念很有用，而且由于 select 是很常见、可移植的 API。但是它也有一些严重的缺陷，在监视的文件描述符非常大的时候就会出现。有限的文件描述符的集合大小。糟糕的性能。从文件描述符的大小开始。FD_SETSIZE 是一个编译期常数，在如今的操作系统中，它的值通常是 1024。它被硬编码在 glibc 的头文件里，并且不容易修改。它把 select 能够监视的文件描述符的数量限制在 1024 以内。曾有些人想要写出能够处理上万个并发访问的客户端请求的服务器，所以这个问题很有现实意义。有一些方法，但是不可移植，也很难用。糟糕的性能问题就好解决的多，但是依然非常严重。注意当 select 返回的时候，它向调用者提供的信息是 “就绪的” 描述符的个数，还有被修改过的描述符集合。描述符集映射着描述符“就绪/未就绪”，但是并没有提供什么有效的方法去遍历所有就绪的描述符。如果只有一个描述符是就绪的，最坏的情况是调用者需要遍历 整个集合 来找到那个描述符。这在监视的描述符数量比较少的时候还行，但是如果数量变的很大的时候，这种方法弊端就凸显出了 注7 。由于这些原因，为了写出高性能的并发服务器， select 已经不怎么用了。每一个流行的操作系统有独特的不可移植的 API，允许用户写出非常高效的事件循环；像框架这样的高级结构还有高级语言通常在一个可移植的接口中包含这些 API。epoll举个例子，来看看 epoll，Linux 上的关于高容量 I/O 事件通知问题的解决方案。epoll 高效的关键之处在于它与内核更好的协作。不是使用文件描述符，epoll_wait 用当前准备好的事件填满一个缓冲区。只有准备好的事件添加到了缓冲区，因此没有必要遍历客户端中当前 所有 监视的文件描述符。这简化了查找就绪的描述符的过程，把空间复杂度从 select 中的 O(N) 变为了 O(1)。关于 epoll API 的完整展示不是这里的目的，网上有很多相关资源。虽然你可能猜到了，我还要写一个不同的并发服务器，这次是用 epool 而不是 select。完整的示例代码 在这里。实际上，由于大部分代码和 用 select 的服务器相同，所以我只会讲要点，在主循环里使用 epoll：struct epoll_event accept_event; accept_event.data.fd = listener_sockfd; accept_event.events = EPOLLIN; if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listener_sockfd, &accept_event) < 0) { perror_die("epoll_ctl EPOLL_CTL_ADD"); } struct epoll_event* events = calloc(MAXFDS, sizeof(struct epoll_event)); if (events == NULL) { die("Unable to allocate memory for epoll_events"); } while (1) { int nready = epoll_wait(epollfd, events, MAXFDS, -1); for (int i = 0; i < nready; i++) { if (events[i].events & EPOLLERR) { perror_die("epoll_wait returned EPOLLERR"); } if (events[i].data.fd == listener_sockfd) { // 监听的套接字就绪了；意味着新客户端正在连接。 ... } else { // A peer socket is ready. if (events[i].events & EPOLLIN) { // 准备好了读取 ... } else if (events[i].events & EPOLLOUT) { // 准备好了写入 ... } } } } 通过调用 epoll_ctl 来配置 epoll。这时，配置监听的套接字数量，也就是 epoll 监听的描述符的数量。然后分配一个缓冲区，把就绪的事件传给 epoll 以供修改。在主循环里对 epoll_wait 的调用是魅力所在。它阻塞着，直到某个描述符就绪了（或者超时），返回就绪的描述符数量。但这时，不要盲目地迭代所有监视的集合，我们知道 epoll_write 会修改传给它的 events 缓冲区，缓冲区中有就绪的事件，从 0 到 nready-1，因此我们只需迭代必要的次数。要在 select 里面重新遍历，有明显的差异：如果在监视着 1000 个描述符，只有两个就绪， epoll_waits 返回的是 nready=2，然后修改 events 缓冲区最前面的两个元素，因此我们只需要“遍历”两个描述符。用 select 我们就需要遍历 1000 个描述符，找出哪个是就绪的。因此，在繁忙的服务器上，有许多活跃的套接字时 epoll 比 select 更加容易扩展。剩下的代码很直观，因为我们已经很熟悉 “select 服务器” 了。实际上，“epoll 服务器” 中的所有“业务逻辑”和 “select 服务器” 是一样的，回调构成相同的代码。这种相似是通过将事件循环抽象分离到一个库/框架中。我将会详述这些内容，因为很多优秀的程序员曾经也是这样做的。相反，下一篇文章里我们会了解 libuv，一个最近出现的更加受欢迎的时间循环抽象层。像 libuv 这样的库让我们能够写出并发的异步服务器，并且不用考虑系统调用下繁琐的细节。注1：我试着在做网络浏览和阅读这两件事的实际差别中突显自己，但经常做得头疼。有很多不同的选项，从“它们是一样的东西”到“一个是另一个的子集”，再到“它们是完全不同的东西”。在面临这样主观的观点时，最好是完全放弃这个问题，专注特殊的例子和用例。注2：POSIX 表示这可以是 EAGAIN，也可以是 EWOULDBLOCK，可移植应用应该对这两个都进行检查。注3：和这个系列所有的 C 示例类似，代码中用到了某些助手工具来设置监听套接字。这些工具的完整代码在这个 仓库 的 utils 模块里。注4：select 不是网络/套接字专用的函数，它可以监视任意的文件描述符，有可能是硬盘文件、管道、终端、套接字或者 Unix 系统中用到的任何文件描述符。这篇文章里，我们主要关注它在套接字方面的应用。注5：有多种方式用多线程来实现事件驱动，我会把它放在稍后的文章中进行讨论。注6：由于各种非实验因素，它 仍然 可以阻塞，即使是在 select 说它就绪了之后。因此服务器上打开的所有套接字都被设置成非阻塞模式，如果对 recv 或 send 的调用返回了 EAGAIN 或者 EWOULDBLOCK，回调函数就装作没有事件发生。阅读示例代码的注释可以了解更多细节。注7：注意这比该文章前面所讲的异步轮询的例子要稍好一点。轮询需要 一直 发生，而 select 实际上会阻塞到有一个或多个套接字准备好读取/写入；select 会比一直询问浪费少得多的 CPU 时间。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-3-event-driven/作者：Eli Bendersky 译者：GitFuture 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是并发网络服务器系列的第二节。第一节 提出了服务端实现的协议，还有简单的顺序服务器的代码，是这整个系列的基础。这一节里，我们来看看怎么用多线程来实现并发，用 C 实现一个最简单的多线程服务器，和用 Python 实现的线程池。该系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动多线程的方法设计并发服务器说起第一节里的顺序服务器的性能，最显而易见的，是在服务器处理客户端连接时，计算机的很多资源都被浪费掉了。尽管假定客户端快速发送完消息，不做任何等待，仍然需要考虑网络通信的开销；网络要比现在的 CPU 慢上百万倍还不止，因此 CPU 运行服务器时会等待接收套接字的流量，而大量的时间都花在完全不必要的等待中。这里是一份示意图，表明顺序时客户端的运行过程：这个图片上有 3 个客户端程序。棱形表示客户端的“到达时间”（即客户端尝试连接服务器的时间）。黑色线条表示“等待时间”（客户端等待服务器真正接受连接所用的时间），有色矩形表示“处理时间”（服务器和客户端使用协议进行交互所用的时间）。有色矩形的末端表示客户端断开连接。上图中，绿色和橘色的客户端尽管紧跟在蓝色客户端之后到达服务器，也要等到服务器处理完蓝色客户端的请求。这时绿色客户端得到响应，橘色的还要等待一段时间。多线程服务器会开启多个控制线程，让操作系统管理 CPU 的并发（使用多个 CPU 核心）。当客户端连接的时候，创建一个线程与之交互，而在主线程中，服务器能够接受其他的客户端连接。下图是该模式的时间轴：每个客户端一个线程，在 C 语言里要用 pthread这篇文章的 第一个示例代码 是一个简单的 “每个客户端一个线程” 的服务器，用 C 语言编写，使用了 phtreads API 用于实现多线程。这里是主循环代码：while (1) { struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); pthread_t the_thread; thread_config_t* config = (thread_config_t*)malloc(sizeof(*config)); if (!config) { die("OOM"); } config->sockfd = newsockfd; pthread_create(&the_thread, NULL, server_thread, config); // 回收线程 —— 在线程结束的时候，它占用的资源会被回收 // 因为主线程在一直运行，所以它比服务线程存活更久。 pthread_detach(the_thread); } 这是 server_thread 函数：void* server_thread(void* arg) { thread_config_t* config = (thread_config_t*)arg; int sockfd = config->sockfd; free(config); // This cast will work for Linux, but in general casting pthread_id to an 这个类型转换在 Linux 中可以正常运行，但是一般来说将 pthread_id 类型转换成整形不便于移植代码 // integral type isn't portable. unsigned long id = (unsigned long)pthread_self(); printf("Thread %lu created to handle connection with socket %d\n", id, sockfd); serve_connection(sockfd); printf("Thread %lu done\n", id); return 0; } 线程 “configuration” 是作为 thread_config_t 结构体进行传递的：typedef struct { int sockfd; } thread_config_t; 主循环中调用的 pthread_create 产生一个新线程，然后运行 server_thread 函数。这个线程会在 server_thread 返回的时候结束。而在 serve_connection 返回的时候 server_thread 才会返回。serve_connection 和第一节完全一样。第一节中我们用脚本生成了多个并发访问的客户端，观察服务器是怎么处理的。现在来看看多线程服务器的处理结果：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-20 06:31:56,632:conn1 connected... INFO:2017-09-20 06:31:56,632:conn2 connected... INFO:2017-09-20 06:31:56,632:conn0 connected... INFO:2017-09-20 06:31:56,632:conn1 sending b'^abc$de^abte$f' INFO:2017-09-20 06:31:56,632:conn2 sending b'^abc$de^abte$f' INFO:2017-09-20 06:31:56,632:conn0 sending b'^abc$de^abte$f' INFO:2017-09-20 06:31:56,633:conn1 received b'b' INFO:2017-09-20 06:31:56,633:conn2 received b'b' INFO:2017-09-20 06:31:56,633:conn0 received b'b' INFO:2017-09-20 06:31:56,670:conn1 received b'cdbcuf' INFO:2017-09-20 06:31:56,671:conn0 received b'cdbcuf' INFO:2017-09-20 06:31:56,671:conn2 received b'cdbcuf' INFO:2017-09-20 06:31:57,634:conn1 sending b'xyz^123' INFO:2017-09-20 06:31:57,634:conn2 sending b'xyz^123' INFO:2017-09-20 06:31:57,634:conn1 received b'234' INFO:2017-09-20 06:31:57,634:conn0 sending b'xyz^123' INFO:2017-09-20 06:31:57,634:conn2 received b'234' INFO:2017-09-20 06:31:57,634:conn0 received b'234' INFO:2017-09-20 06:31:58,635:conn1 sending b'25$^ab0000$abab' INFO:2017-09-20 06:31:58,635:conn2 sending b'25$^ab0000$abab' INFO:2017-09-20 06:31:58,636:conn1 received b'36bc1111' INFO:2017-09-20 06:31:58,636:conn2 received b'36bc1111' INFO:2017-09-20 06:31:58,637:conn0 sending b'25$^ab0000$abab' INFO:2017-09-20 06:31:58,637:conn0 received b'36bc1111' INFO:2017-09-20 06:31:58,836:conn2 disconnecting INFO:2017-09-20 06:31:58,836:conn1 disconnecting INFO:2017-09-20 06:31:58,837:conn0 disconnecting 实际上，所有客户端同时连接，它们与服务器的通信是同时发生的。每个客户端一个线程的难点尽管在现代操作系统中就资源利用率方面来看，线程相当的高效，但前一节中讲到的方法在高负载时却会出现纰漏。想象一下这样的情景：很多客户端同时进行连接，某些会话持续的时间长。这意味着某个时刻服务器上有很多活跃的线程。太多的线程会消耗掉大量的内存和 CPU 资源，而仅仅是用于上下文切换 注1 。另外其也可视为安全问题：因为这样的设计容易让服务器成为 DoS 攻击 的目标 —— 上百万个客户端同时连接，并且客户端都处于闲置状态，这样耗尽了所有资源就可能让服务器宕机。当服务器要与每个客户端通信，CPU 进行大量计算时，就会出现更严重的问题。这种情况下，容易想到的方法是减少服务器的响应能力 —— 只有其中一些客户端能得到服务器的响应。因此，对多线程服务器所能够处理的并发客户端数做一些 速率限制 就是个明智的选择。有很多方法可以实现。最容易想到的是计数当前已经连接上的客户端，把连接数限制在某个范围内（需要通过仔细的测试后决定）。另一种流行的多线程应用设计是使用 线程池。线程池线程池 很简单，也很有用。服务器创建几个任务线程，这些线程从某些队列中获取任务。这就是“池”。然后每一个客户端的连接被当成任务分发到池中。只要池中有空闲的线程，它就会去处理任务。如果当前池中所有线程都是繁忙状态，那么服务器就会阻塞，直到线程池可以接受任务（某个繁忙状态的线程处理完当前任务后，变回空闲的状态）。这里有个 4 线程的线程池处理任务的图。任务（这里就是客户端的连接）要等到线程池中的某个线程可以接受新任务。非常明显，线程池的定义就是一种按比例限制的机制。我们可以提前设定服务器所能拥有的线程数。那么这就是并发连接的最多的客户端数 —— 其它的客户端就要等到线程空闲。如果我们的池中有 8 个线程，那么 8 就是服务器可以处理的最多的客户端并发连接数，哪怕上千个客户端想要同时连接。那么怎么确定池中需要有多少个线程呢？通过对问题范畴进行细致的分析、评估、实验以及根据我们拥有的硬件配置。如果是单核的云服务器，答案只有一个；如果是 100 核心的多套接字的服务器，那么答案就有很多种。也可以在运行时根据负载动态选择池的大小 —— 我会在这个系列之后的文章中谈到这个东西。使用线程池的服务器在高负载情况下表现出 性能退化 —— 客户端能够以稳定的速率进行连接，可能会比其它时刻得到响应的用时稍微久一点；也就是说，无论多少个客户端同时进行连接，服务器总能保持响应，尽最大能力响应等待的客户端。与之相反，每个客户端一个线程的服务器，会接收多个客户端的连接直到过载，这时它更容易崩溃或者因为要处理所有客户端而变得缓慢，因为资源都被耗尽了（比如虚拟内存的占用）。在服务器上使用线程池为了改变服务器的实现，我用了 Python，在 Python 的标准库中带有一个已经实现好的稳定的线程池。（concurrent.futures 模块里的 ThreadPoolExecutor） 注2 。服务器创建一个线程池，然后进入循环，监听套接字接收客户端的连接。用 submit 把每一个连接的客户端分配到池中：pool = ThreadPoolExecutor(args.n) sockobj = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sockobj.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sockobj.bind(('localhost', args.port)) sockobj.listen(15) try: while True: client_socket, client_address = sockobj.accept() pool.submit(serve_connection, client_socket, client_address) except KeyboardInterrupt as e: print(e) sockobj.close() serve_connection 函数和 C 的那部分很像，与一个客户端交互，直到其断开连接，并且遵循我们的协议：ProcessingState = Enum('ProcessingState', 'WAIT_FOR_MSG IN_MSG') def serve_connection(sockobj, client_address): print('{0} connected'.format(client_address)) sockobj.sendall(b'*') state = ProcessingState.WAIT_FOR_MSG while True: try: buf = sockobj.recv(1024) if not buf: break except IOError as e: break for b in buf: if state == ProcessingState.WAIT_FOR_MSG: if b == ord(b'^'): state = ProcessingState.IN_MSG elif state == ProcessingState.IN_MSG: if b == ord(b'$'): state = ProcessingState.WAIT_FOR_MSG else: sockobj.send(bytes([b + 1])) else: assert False print('{0} done'.format(client_address)) sys.stdout.flush() sockobj.close() 来看看线程池的大小对并行访问的客户端的阻塞行为有什么样的影响。为了演示，我会运行一个池大小为 2 的线程池服务器（只生成两个线程用于响应客户端）。$ python3.6 threadpool-server.py -n 2 在另外一个终端里，运行客户端模拟器，产生 3 个并发访问的客户端：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-22 05:58:52,815:conn1 connected... INFO:2017-09-22 05:58:52,827:conn0 connected... INFO:2017-09-22 05:58:52,828:conn1 sending b'^abc$de^abte$f' INFO:2017-09-22 05:58:52,828:conn0 sending b'^abc$de^abte$f' INFO:2017-09-22 05:58:52,828:conn1 received b'b' INFO:2017-09-22 05:58:52,828:conn0 received b'b' INFO:2017-09-22 05:58:52,867:conn1 received b'cdbcuf' INFO:2017-09-22 05:58:52,867:conn0 received b'cdbcuf' INFO:2017-09-22 05:58:53,829:conn1 sending b'xyz^123' INFO:2017-09-22 05:58:53,829:conn0 sending b'xyz^123' INFO:2017-09-22 05:58:53,830:conn1 received b'234' INFO:2017-09-22 05:58:53,831:conn0 received b'2' INFO:2017-09-22 05:58:53,831:conn0 received b'34' INFO:2017-09-22 05:58:54,831:conn1 sending b'25$^ab0000$abab' INFO:2017-09-22 05:58:54,832:conn1 received b'36bc1111' INFO:2017-09-22 05:58:54,832:conn0 sending b'25$^ab0000$abab' INFO:2017-09-22 05:58:54,833:conn0 received b'36bc1111' INFO:2017-09-22 05:58:55,032:conn1 disconnecting INFO:2017-09-22 05:58:55,032:conn2 connected... INFO:2017-09-22 05:58:55,033:conn2 sending b'^abc$de^abte$f' INFO:2017-09-22 05:58:55,033:conn0 disconnecting INFO:2017-09-22 05:58:55,034:conn2 received b'b' INFO:2017-09-22 05:58:55,071:conn2 received b'cdbcuf' INFO:2017-09-22 05:58:56,036:conn2 sending b'xyz^123' INFO:2017-09-22 05:58:56,036:conn2 received b'234' INFO:2017-09-22 05:58:57,037:conn2 sending b'25$^ab0000$abab' INFO:2017-09-22 05:58:57,038:conn2 received b'36bc1111' INFO:2017-09-22 05:58:57,238:conn2 disconnecting 回顾之前讨论的服务器行为：在顺序服务器中，所有的连接都是串行的。一个连接结束后，下一个连接才能开始。前面讲到的每个客户端一个线程的服务器中，所有连接都被同时接受并得到服务。这里可以看到一种可能的情况：两个连接同时得到服务，只有其中一个结束连接后第三个才能连接上。这就是把线程池大小设置成 2 的结果。真实用例中我们会把线程池设置的更大些，取决于机器和实际的协议。线程池的缓冲机制就能很好理解了 —— 我 几个月前 更详细的介绍过这种机制，关于 Clojure 的 core.async 模块。总结与展望这篇文章讨论了在服务器中，用多线程作并发的方法。每个客户端一个线程的方法最早提出来，但是实际上却不常用，因为它并不安全。线程池就常见多了，最受欢迎的几个编程语言有良好的实现（某些编程语言，像 Python，就是在标准库中实现）。这里说的使用线程池的服务器，不会受到每个客户端一个线程的弊端。然而，线程不是处理多个客户端并行访问的唯一方法。下一节中我们会看看其它的解决方案，可以使用异步处理，或者事件驱动的编程。注1：老实说，现代 Linux 内核可以承受足够多的并发线程 —— 只要这些线程主要在 I/O 上被阻塞。这里有个示例程序，它产生可配置数量的线程，线程在循环体中是休眠的，每 50 ms 唤醒一次。我在 4 核的 Linux 机器上可以轻松的产生 10000 个线程；哪怕这些线程大多数时间都在睡眠，它们仍然消耗一到两个核心，以便实现上下文切换。而且，它们占用了 80 GB 的虚拟内存（Linux 上每个线程的栈大小默认是 8MB）。实际使用中，线程会使用内存并且不会在循环体中休眠，因此它可以非常快的占用完一个机器的内存。注2：自己动手实现一个线程池是个有意思的练习，但我现在还不想做。我曾写过用来练手的 针对特殊任务的线程池。是用 Python 写的；用 C 重写的话有些难度，但对于经验丰富的程序员，几个小时就够了。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-2-threads/作者：Eli Bendersky 译者：GitFuture 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是关于并发网络服务器编程的第一篇教程。我计划测试几个主流的、可以同时处理多个客户端请求的服务器并发模型，基于可扩展性和易实现性对这些模型进行评判。所有的服务器都会监听套接字连接，并且实现一些简单的协议用于与客户端进行通讯。该系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动协议该系列教程所用的协议都非常简单，但足以展示并发服务器设计的许多有趣层面。而且这个协议是 有状态的 —— 服务器根据客户端发送的数据改变内部状态，然后根据内部状态产生相应的行为。并非所有的协议都是有状态的 —— 实际上，基于 HTTP 的许多协议是无状态的，但是有状态的协议也是很常见，值得认真讨论。在服务器端看来，这个协议的视图是这样的：总之：服务器等待新客户端的连接；当一个客户端连接的时候，服务器会向该客户端发送一个 * 字符，进入“等待消息”的状态。在该状态下，服务器会忽略客户端发送的所有字符，除非它看到了一个 ^ 字符，这表示一个新消息的开始。这个时候服务器就会转变为“正在通信”的状态，这时它会向客户端回送数据，把收到的所有字符的每个字节加 1 回送给客户端 注1 。当客户端发送了 $ 字符，服务器就会退回到等待新消息的状态。^ 和 $ 字符仅仅用于分隔消息 —— 它们不会被服务器回送。每个状态之后都有个隐藏的箭头指向 “等待客户端” 状态，用于客户端断开连接。因此，客户端要表示“我已经结束”的方法很简单，关掉它那一端的连接就好。显然，这个协议是真实协议的简化版，真实使用的协议一般包含复杂的报文头、转义字符序列（例如让消息体中可以出现 $ 符号），额外的状态变化。但是我们这个协议足以完成期望。另一点：这个系列是介绍性的，并假设客户端都工作的很好（虽然可能运行很慢）；因此没有设置超时，也没有设置特殊的规则来确保服务器不会因为客户端的恶意行为（或是故障）而出现阻塞，导致不能正常结束。顺序服务器这个系列中我们的第一个服务端程序是一个简单的“顺序”服务器，用 C 进行编写，除了标准的 POSIX 中用于套接字的内容以外没有使用其它库。服务器程序是顺序，因为它一次只能处理一个客户端的请求；当有客户端连接时，像之前所说的那样，服务器会进入到状态机中，并且不再监听套接字接受新的客户端连接，直到当前的客户端结束连接。显然这不是并发的，而且即便在很少的负载下也不能服务多个客户端，但它对于我们的讨论很有用，因为我们需要的是一个易于理解的基础。这个服务器的完整代码在这里；接下来，我会着重于一些重点的部分。main 函数里面的外层循环用于监听套接字，以便接受新客户端的连接。一旦有客户端进行连接，就会调用 serve_connection，这个函数中的代码会一直运行，直到客户端断开连接。顺序服务器在循环里调用 accept 用来监听套接字，并接受新连接：while (1) { struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); serve_connection(newsockfd); printf("peer done\n"); } accept 函数每次都会返回一个新的已连接的套接字，然后服务器调用 serve_connection；注意这是一个 阻塞式 的调用 —— 在 serve_connection 返回前，accept 函数都不会再被调用了；服务器会被阻塞，直到客户端结束连接才能接受新的连接。换句话说，客户端按 顺序 得到响应。这是 serve_connection 函数：typedef enum { WAIT_FOR_MSG, IN_MSG } ProcessingState; void serve_connection(int sockfd) { if (send(sockfd, "*", 1, 0) < 1) { perror_die("send"); } ProcessingState state = WAIT_FOR_MSG; while (1) { uint8_t buf[1024]; int len = recv(sockfd, buf, sizeof buf, 0); if (len < 0) { perror_die("recv"); } else if (len == 0) { break; } for (int i = 0; i < len; ++i) { switch (state) { case WAIT_FOR_MSG: if (buf[i] == '^') { state = IN_MSG; } break; case IN_MSG: if (buf[i] == '$') { state = WAIT_FOR_MSG; } else { buf[i] += 1; if (send(sockfd, &buf[i], 1, 0) < 1) { perror("send error"); close(sockfd); return; } } break; } } } close(sockfd); } 它完全是按照状态机协议进行编写的。每次循环的时候，服务器尝试接收客户端的数据。收到 0 字节意味着客户端断开连接，然后循环就会退出。否则，会逐字节检查接收缓存，每一个字节都可能会触发一个状态。recv 函数返回接收到的字节数与客户端发送消息的数量完全无关（^...$ 闭合序列的字节）。因此，在保持状态的循环中遍历整个缓冲区很重要。而且，每一个接收到的缓冲中可能包含多条信息，但也有可能开始了一个新消息，却没有显式的结束字符；而这个结束字符可能在下一个缓冲中才能收到，这就是处理状态在循环迭代中进行维护的原因。例如，试想主循环中的 recv 函数在某次连接中返回了三个非空的缓冲：^abc$de^abte$fxyz^12325$^ab$abab服务端返回的是哪些数据？追踪代码对于理解状态转变很有用。（答案见 注 2 ）多个并发客户端如果多个客户端在同一时刻向顺序服务器发起连接会发生什么事情？服务器端的代码（以及它的名字 “顺序服务器”）已经说的很清楚了，一次只能处理 一个 客户端的请求。只要服务器在 serve_connection 函数中忙于处理客户端的请求，就不会接受别的客户端的连接。只有当前的客户端断开了连接，serve_connection 才会返回，然后最外层的循环才能继续执行接受其他客户端的连接。为了演示这个行为，该系列教程的示例代码 包含了一个 Python 脚本，用于模拟几个想要同时连接服务器的客户端。每一个客户端发送类似之前那样的三个数据缓冲 注3 ，不过每次发送数据之间会有一定延迟。客户端脚本在不同的线程中并发地模拟客户端行为。这是我们的序列化服务器与客户端交互的信息记录：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-16 14:14:17,763:conn1 connected... INFO:2017-09-16 14:14:17,763:conn1 sending b'^abc$de^abte$f' INFO:2017-09-16 14:14:17,763:conn1 received b'b' INFO:2017-09-16 14:14:17,802:conn1 received b'cdbcuf' INFO:2017-09-16 14:14:18,764:conn1 sending b'xyz^123' INFO:2017-09-16 14:14:18,764:conn1 received b'234' INFO:2017-09-16 14:14:19,764:conn1 sending b'25$^ab0000$abab' INFO:2017-09-16 14:14:19,765:conn1 received b'36bc1111' INFO:2017-09-16 14:14:19,965:conn1 disconnecting INFO:2017-09-16 14:14:19,966:conn2 connected... INFO:2017-09-16 14:14:19,967:conn2 sending b'^abc$de^abte$f' INFO:2017-09-16 14:14:19,967:conn2 received b'b' INFO:2017-09-16 14:14:20,006:conn2 received b'cdbcuf' INFO:2017-09-16 14:14:20,968:conn2 sending b'xyz^123' INFO:2017-09-16 14:14:20,969:conn2 received b'234' INFO:2017-09-16 14:14:21,970:conn2 sending b'25$^ab0000$abab' INFO:2017-09-16 14:14:21,970:conn2 received b'36bc1111' INFO:2017-09-16 14:14:22,171:conn2 disconnecting INFO:2017-09-16 14:14:22,171:conn0 connected... INFO:2017-09-16 14:14:22,172:conn0 sending b'^abc$de^abte$f' INFO:2017-09-16 14:14:22,172:conn0 received b'b' INFO:2017-09-16 14:14:22,210:conn0 received b'cdbcuf' INFO:2017-09-16 14:14:23,173:conn0 sending b'xyz^123' INFO:2017-09-16 14:14:23,174:conn0 received b'234' INFO:2017-09-16 14:14:24,175:conn0 sending b'25$^ab0000$abab' INFO:2017-09-16 14:14:24,176:conn0 received b'36bc1111' INFO:2017-09-16 14:14:24,376:conn0 disconnecting 这里要注意连接名：conn1 是第一个连接到服务器的，先跟服务器交互了一段时间。接下来的连接 conn2 —— 在第一个断开连接后，连接到了服务器，然后第三个连接也是一样。就像日志显示的那样，每一个连接让服务器变得繁忙，持续了大约 2.2 秒的时间（这实际上是人为地在客户端代码中加入的延迟），在这段时间里别的客户端都不能连接。显然，这不是一个可扩展的策略。这个例子中，客户端中加入了延迟，让服务器不能处理别的交互动作。一个智能服务器应该能处理一堆客户端的请求，而这个原始的服务器在结束连接之前一直繁忙（我们将会在之后的章节中看到如何实现智能的服务器）。尽管服务端有延迟，但这不会过度占用 CPU；例如，从数据库中查找信息（时间基本上是花在连接到数据库服务器上，或者是花在硬盘中的本地数据库）。总结及期望这个示例服务器达成了两个预期目标：首先是介绍了问题范畴和贯彻该系列文章的套接字编程基础。对于并发服务器编程的抛砖引玉 —— 就像之前的部分所说，顺序服务器还不能在非常轻微的负载下进行扩展，而且没有高效的利用资源。在看下一篇文章前，确保你已经理解了这里所讲的服务器/客户端协议，还有顺序服务器的代码。我之前介绍过了这个简单的协议；例如 串行通信分帧 和 用协程来替代状态机。要学习套接字网络编程的基础，Beej 的教程 用来入门很不错，但是要深入理解我推荐你还是看本书。如果有什么不清楚的，请在评论区下进行评论或者向我发送邮件。深入理解并发服务器！注1：状态转变中的 In/Out 记号是指 Mealy machine。注2：回应的是 bcdbcuf23436bc。注3：这里在结尾处有一点小区别，加了字符串 0000 —— 服务器回应这个序列，告诉客户端让其断开连接；这是一个简单的握手协议，确保客户端有足够的时间接收到服务器发送的所有回复。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-1-introduction/作者：Eli Bendersky 译者：GitFuture 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
最近我开始发力钻研 Python 的新 asyncio 模块。原因是我需要做一些事情，使用事件 IO 会使这些事情工作得更好，炙手可热的 asynio 正好可以用来牛刀小试。从试用的经历来看，该模块比我预想的复杂许多，我现在可以非常肯定地说，我不知道该如何恰当地使用 asyncio。从 Twisted 框架借鉴一些经验来理解 asynio 并非难事，但是，asyncio 包含众多的元素，我开始动摇，不知道如何将这些孤立的零碎拼图组合成一副完整的图画。我已没有足够的智力提出任何更好的建议，在这里，只想分享我的困惑，求大神指点。原语asyncio 通过 协程 coroutines 的帮助来实现异步 IO。最初它是通过 yield 和 yield from 表达式实现的一个库，因为 Python 语言本身演进的缘故，现在它已经变成一个更复杂的怪兽。所以，为了在同一个频道讨论下去，你需要了解如下一些术语：事件循环事件循环策略awaitable协程函数老式协程函数协程协程封装 生成器 （ generator ） future并发的future 任务 （ task ） 句柄 执行器 （ executor ） 传输 （ transport ） 协议此外，Python 还新增了一些新的特殊方法：__aenter__ 和 __aenter__，用于异步块操作__aiter__ 和 __anext__，用于异步迭代器（异步循环和异步推导）。为了更强大些，协议已经改变过一次了。 在 Python 3.5 它返回一个 awaitable（这是个协程）；在 3.6它返回一个新的异步生成器。__await__，用于自定义的 awaitable你还需要了解相当多的内容，文档涵盖了那些部分。尽管如此，我做了一些额外说明以便对其有更好的理解：事件循环asyncio 事件循环和你第一眼看上去的略有不同。表面看，每个线程都有一个事件循环，然而事实并非如此。我认为它们应该按照如下的方式工作：如果是主线程，当调用 asyncio.get_event_loop() 时创建一个事件循环。如果是其它线程，当调用 asyncio.get_event_loop() 时返回运行时错误。当前线程可以使用 asyncio.set_event_loop() 在任何时间节点绑定事件循环。该事件循环可由 asyncio.new_evet_loop() 函数创建。事件循环可以在不绑定到当前线程的情况下使用。asyncio.get_event_loop() 返回绑定线程的事件循环，而非当前运行的事件循环。这些行为的组合是超混淆的，主要有以下几个原因。 首先，你需要知道这些函数被委托到全局设置的底层事件循环策略。 默认是将事件循环绑定到线程。 或者，如果需要的话，可以在理论上将事件循环绑定到一个 greenlet 或类似的。 然而，重要的是要知道库代码不控制策略，因此不能推断 asyncio 将适用于线程。其次，asyncio 不需要通过策略将事件循环绑定到上下文。 事件循环可以单独工作。 但是这正是库代码的第一个问题，因为协同程序或类似的东西并不知道哪个事件循环负责调度它。 这意味着，如果从协程中调用 asyncio.get_event_loop()，你可能没有机会取得事件循环。 这也是所有 API 均采用可选的显式事件循环参数的原因。 举例来说，要弄清楚当前哪个协程正在运行，不能使用如下调用：def get_task(): loop = asyncio.get_event_loop() try: return asyncio.Task.get_current(loop) except RuntimeError: return None 相反，必须显式地传递事件循环。 这进一步要求你在库代码中显式地遍历事件循环，否则可能发生很奇怪的事情。 我不知道这种设计的思想是什么，但如果不解决这个问题（例如 get_event_loop() 返回实际运行的事件循环），那么唯一有意义的其它方案是明确禁止显式事件循环传递，并要求它绑定到当前上下文（线程等）。由于事件循环策略不提供当前上下文的标识符，因此库也不可能以任何方式“索引”到当前上下文。 也没有回调函数用来监视这样的上下文的拆除，这进一步限制了实际可以开展的操作。awaitable 与 协程 coroutine 以我的愚见，Python 最大的设计错误是过度重载迭代器。它们现在不仅用于迭代，而且用于各种类型的协程。 Python 中迭代器最大的设计错误之一是如果 StopIteration 没有被捕获形成的空泡。 这可能导致非常令人沮丧的问题，其中某处的异常可能导致其它地方的生成器或协同程序中止。 这是一个长期存在的问题，基于 Python 的模板引擎如 Jinja 经常面临这种问题。 该模板引擎在内部渲染为生成器，并且当由于某种原因的模板引起 StopIteration 时，渲染就停止在那里。Python 慢慢认识到了过度重载的教训。 首先在 3.x 版本加入 asyncio 模块，并没有语言级支持。 所以自始至终它不过仅仅是装饰器和生成器而已。 为了实现 yield from 以及其它东西，StopIteration 再次重载。 这导致了令人困惑的行为，像这样：>>> def foo(n): ... if n in (0, 1): ... return [1] ... for item in range(n): ... yield item * 2 ... >>> list(foo(0)) [] >>> list(foo(1)) [] >>> list(foo(2)) [0, 2] 没有错误，没有警告。只是不是你所期望的行为。 这是因为从一个作为生成器的函数中 return 的值实际上引发了一个带有单个参数的 StopIteration，它不是由迭代器协议捕获的，而只是在协程代码中处理。在 3.5 和 3.6 有很多改变，因为现在除了生成器我们还有协程对象。除了通过封装生成器来生成协程，没有其它可以直接生成协程的单独对象。它是通过用给函数加 async 前缀来实现。 例如 async def x() 会产生这样的协程。 现在在 3.6，将有单独的异步生成器，它通过触发 AsyncStopIteration 保持其独立性。 此外，对于Python 3.5 和更高版本，导入新的 future 对象（generator_stop），如果代码在迭代步骤中触发 StopIteration，它将引发 RuntimeError。为什么我提到这一切？ 因为老的实现方式并未真的消失。 生成器仍然具有 send 和 throw 方法以及协程仍然在很大程度上表现为生成器。你需要知道这些东西，它们将在未来伴随你相当长的时间。为了统一很多这样的重复，现在我们在 Python 中有更多的概念了：awaitable：具有__await__方法的对象。 由本地协同程序和旧式协同程序以及一些其它程序实现。 协程函数 （ coroutinefunction ） ：返回原生协程的函数。 不要与返回协程的函数混淆。 协程 （ coroutine ） : 原生的协程程序。 注意，目前为止，当前文档不认为老式 asyncio 协程是协程程序。 至少 inspect.iscoroutine 不认为它是协程。 尽管它被 future/awaitable 分支接纳。特别令人困惑的是 asyncio.iscoroutinefunction 和inspect.iscoroutinefunction 正在做不同的事情，这与 inspect.iscoroutine 和 inspect.iscoroutinefunction 情况相同。 值得注意的是，尽管 inspect 在类型检查中不知道有关 asycnio 旧式协程函数的任何信息，但是当您检查 awaitable 状态时它显然知道它们，即使它与 **await** 不一致。 协程封装器 coroutine wrapper 每当你运行 async def ，Python 就会调用一个线程局部的协程封装器。它由 sys.set_coroutine_wrapper 设置，并且它是可以包装这些东西的一个函数。 看起来有点像如下代码：>>> import sys >>> sys.set_coroutine_wrapper(lambda x: 42) >>> async def foo(): ... pass ... >>> foo() __main__:1: RuntimeWarning: coroutine 'foo' was never awaited 42 在这种情况下，我从来没有实际调用原始的函数，只是给你一个提示，说明这个函数可以做什么。 目前我只能说它总是线程局部有效，所以，如果替换事件循环策略，你需要搞清楚如何让协程封装器在相同的上下文同步更新。创建的新线程不会从父线程继承那些标识。这不要与 asyncio 协程封装代码混淆。awaitable 和 future有些东西是 awaitable 的。 据我所见，以下概念被认为是 awaitable：原生的协程配置了假的 CO_ITERABLE_COROUTINE 标识的生成器（文中有涉及）具有 __await__ 方法的对象除了生成器由于历史遗留的原因不使用之外，其它的对象都使用 __await__ 方法。 CO_ITERABLE_COROUTINE 标志来自哪里？它来自一个协程封装器（现在与 sys.set_coroutine_wrapper 有些混淆），即 @asyncio.coroutine。 通过一些间接方法，它使用 types.coroutine（现在与 types.CoroutineType 或 asyncio.coroutine 有些混淆）封装生成器，并通过另外一个标志 CO_ITERABLE_COROUTINE 重新创建内部代码对象。所以既然我们知道这些东西是什么，那么什么是 future？ 首先，我们需要澄清一件事情：在 Python 3 中，实际上有两种（完全不兼容）的 future 类型：asyncio.futures.Future 和 concurrent.futures.Future。 其中一个出现在另一个之前，但它们都仍然在 asyncio 中使用。 例如，asyncio.run_coroutine_threadsafe() 将调度一个协程到在另一个线程中运行的事件循环，但它返回一个 concurrent.futures.Future 对象，而不是 asyncio.futures.Future 对象。 这是有道理的，因为只有 concurrent.futures.Future 对象是线程安全的。所以现在我们知道有两个不兼容的 future，我们应该澄清哪个 future 在 asyncio 中。 老实说，我不完全确定差异在哪里，但我打算暂时称之为“最终”。它是一个最终将持有一个值的对象，当还在计算时你可以对最终结果做一些处理。 future 对象的一些变种称为 deferred，还有一些叫做 promise。 我实在难以理解它们真正的区别。你能用一个 future 对象做什么？ 你可以关联一个准备就绪时将被调用的回调函数，或者你可以关联一个 future 失败时将被触发的回调函数。 此外，你可以 await 它（它实现__await__，因此可等待），此外，future 也可以取消。那么你怎样才能得到这样的 future 对象？ 通过在 awaitable 对象上调用 asyncio.ensure_future。它会把一个旧版的生成器转变为 future 对象。 然而，如果你阅读文档，你会读到 asyncio.ensure_future 实际上返回一个task（任务）。 那么问题来了，什么是任务？任务 任务 （ task ） 某种意义上是一个封装了协程的 futur 对象。它的工作方式和 future 类似，但它也有一些额外的方法来提取所包含的协程的当前堆栈。 我们已经见过了在前面提到过的任务，因为它是通过 Task.get_current 确定事件循环当前正在做什么的主要方式。在如何取消工作方面，任务和 future 也有区别，但这超出了本文的范围。“取消”是它们自己最大的问题。 如果你处于一个协程中，并且知道自己正在运行，你可以通过前面提到的 Task.get_current 获取自己的任务，但这需要你知道自己被派遣在哪个事件循环，该事件循环可能是、也可能不是已绑定的那个线程。协程不可能知道它与哪个循环一起使用。task 也没有提供该信息的公共 API。 然而，如果你确实可以获得一个任务，你可以访问 task._loop，通过它反指到事件循环。句柄除了上面提到的所有一切还有句柄。 句柄是等待执行的不透明对象，不可等待，但可以被取消。 特别是如果你使用 call_soon 或者 call_soon_threadsafe（还有其它一些）调度执行一个调用，你可以获得句柄，然后使用它尽力尝试取消执行，但不能等待实际调用生效。 执行器 Executor 因为你可以有多个事件循环，但这并不意味着每个线程理所当然地应用多个事件循环，最常见的情形还是一个线程一个事件循环。 那么你如何通知另一个事件循环做一些工作？ 你不能到另一个线程的事件循环中执行回调函数并获取结果。 这种情况下，你需要使用执行器。 执行器 Executor 来自 concurrent.futures，它允许你将工作安排到本身未发生事件的线程中。 例如，如果在事件循环中使用 run_in_executor 来调度将在另一个线程中调用的函数。 其返回结果是 asyncio 协程，而不是像 run_coroutine_threadsafe 这样的并发协程。 我还没有足够的心智来弄清楚为什么设计这样的 API，应该如何使用，以及什么时候使用。 文档中建议执行器可以用于构建多进程。传输和协议我总是认为传输与协议也凌乱不堪，实际这部分内容基本上是对 Twisted 的逐字拷贝。详情毋庸赘述，请直接阅读相关文档。如何使用 asyncio现在我们已经大致了解 asyncio，我发现了一些模式，人们似乎在写 asyncio 代码时使用：将事件循环传递给所有协程。 这似乎是社区中一部分人的做法。 把事件循环信息提供给协程为协程获取自己运行的任务提供了可能性。或者你要求事件循环绑定到线程，这也能达到同样的目的。 理想情况下两者都支持。 可悲的是，社区已经分化。如果想使用上下文数据（如线程本地数据），你可谓是运气不佳。 最流行的变通方法显然是 atlassian 的 aiolocals，它基本上需要你手动传递上下文信息到协程，因为解释器不为此提供支持。 这意味着如果你用一个工具类库生成协程，你将失去上下文。忽略 Python 中的旧式协程。 只使用 3.5 版本中 async def 关键字和协程。 你总可能要用到它们，因为在老版本中，没有异步上下文管理器，这是非常必要的资源管理。学习重新启动事件循环进行善后清理。 这部分功能和我预想的不同，我花了比较长的时间来厘清它的实现。清理操作的最好方式是不断重启事件循环直到没有等待事件。 遗憾的是没有什么通用的模式来处理清理操作，你只能用一些丑陋的临时方案糊口度日。 例如 aiohttp 的 web 支持也做这个模式，所以如果你想要结合两个清理逻辑，你可能需要重新实现它提供的工具助手，因为该助手功能实现后，它彻底破坏了事件循环的设计。 当然，它不是我见过的第一个干这种坏事的库 :(。使用子进程是不明显的。 你需要一个事件循环在主线程中运行，我想它是在监听信号事件，然后分派到其它事件循环。 这需要通过 asyncio.get_child_watcher().attach_loop(...) 通知循环。编写同时支持异步和同步的代码在某种程度上注定要失败。 尝试在同一个对象上支持 with 和 async with 是危险的事情。如果你想给一个协程起个更好的名字，弄清楚为什么它没有被等待，设置 __name__没有帮助。 你需要设置 __qualname__ 而不是打印出错误消息来。有时内部类型交换会使你麻痹。 特别是 asyncio.wait() 函数将确保所有的事情都是 future，这意味着如果你传递协程，你将很难发现你的协程是否已经完成或者正在等待，因为输入对象不再匹配输出对象。 在这种情况下，唯一真正理智的做法是确保前期一切都是 future。上下文数据除了疯狂的复杂性和对如何更好地编写 API 缺乏理解，我最大的问题是完全缺乏对上下文本地数据的考虑。这是 Node 社区现在学习的东西。continuation-local-storage 存在，但该实现被接受的太晚。持续本地存储和类似的概念常用于在并发环境中实施安全策略，并且该信息的损坏可能导致严重的安全问题。事实上，Python 甚至没有任何存储，这令人失望至极。我正在研究这个内容，因为我正在调查如何最好地支持 Sentry's breadcrumbs 的 asyncio，然而我并没有看到一个合理的方式做到这一点。在 asyncio 中没有上下文的概念，没有办法从通用代码中找出您正在使用的事件循环，并且如果没有 monkeypatching（运行环境下的补丁），也无法获取这些信息。Node 当前正在经历如何找到这个问题的长期解决方案的过程。这个问题不容忽视，因为它在所有生态系统中反复出现过，如 JavaScript、Python 和 .NET 环境。该问题被命名为异步上下文传播，其解决方案有许多名称。在 Go 中，需要使用上下文包，并明确地传递给所有 goroutine（不是一个完美的解决方案，但至少有一个）。.NET 具有本地调用上下文形式的最佳解决方案。它可以是线程上下文，Web 请求上下文或类似的东西，除非被抑制，否则它会自动传播。微软的解决方案是我们的黄金标准。我现在相信，微软在 15 年前已经解决了该问题。我不知道该生态系统是否还够年轻，还可以添加逻辑调用上下文，可能现在仍然为时未晚。个人感想复杂的东西变得越来越复杂。 我没有随意使用 asyncio 的心智。它需要不断地更新所有 Python 语言的变化的知识，这很大程度上使语言本身变得复杂。 令人鼓舞的是，围绕着它的生态系统正在不断发展，只是不知道还需要几年的时间，才能带给开发者愉快和稳定的开发体验。3.5 版本引入的东西（新的协程对象）非常棒。 特别是这些变化包括引入了一个合理的基础，这些都是我在早期的版本中一直期盼的。在我心中， 通过重载生成器实现协程是一个错误。 关于什么是 asyncio，我难以置喙。 这是一个非常复杂的事情，内部令人眼花缭乱。 我很难理解它工作的所有细节。你什么时候可以传递一个生成器，什么时候它必须是一个真正的协程，future 是什么，任务是什么，事件循环如何工作，这甚至还没有触碰到真正的 IO 部分。最糟糕的是，asyncio 甚至不是特别快。 David Beazley 演示的它设计的 asyncio 的替代品是原生版本速度的两倍。 asyncio 巨复杂，很难理解，也无法兑现自己在主要特性上的承诺，对于它，我只想说我想静静。我知道，至少我对 asyncio 理解的不够透彻，没有足够的信心对人们如何用它构建代码给出建议。作者：Armin Ronacher软件开发者和开源骨灰， Flask 框架的创造者。via: http://lucumr.pocoo.org/2016/10/30/i-dont-understand-asyncio/作者：Armin Ronacher 译者：firstadream 校对：jasminepeng本文由 LCTT 原创编译，Linux中国 荣誉推出