06 | Swoole与Go系列教程之百万协程的应用

yxhsea 的个人博客 / 299 / 1 / 创建于 6个月前 / 更新于 6个月前 / 1 个改进

大家好，我是码农先森。

写在前面

协程的出现是为了解决传统线程和进程模型在并发编程中的一些问题。随着计算机应用的复杂性增加，对于高效处理异步任务的需求也越来越多。传统的线程和进程模型在处理大量的异步任务时会面临资源消耗和切换开销的问题。

在传统的线程或进程模型中，当一个任务发生阻塞时，整个线程或进程都会被阻塞，导致其他任务无法继续执行。这种阻塞会浪费计算资源，影响整体性能。在某些场景下，并不需要使用重量级的线程或进程来进行并发处理。轻量级的协程可以在单个线程中实现多个任务的并发执行，减少了资源的开销。

协程提供了更好的控制权，可以自主地选择在何处暂停和恢复执行，而不是由操作系统的调度器来决定。这种可控制性使得协程更适用于协作式多任务处理。随着多核处理器的普及，仅依靠传统的线程和进程模型无法充分利用多核计算能力。协程可以更好地利用多核处理器，提高并行计算效率。

协程的概念

协程（Coroutine）是一种轻量级的并发编程技术，它可以在单个线程中实现多个控制流的协作执行。与传统的线程或进程相比，协程具有更小的内存占用和更高的执行效率。

协程的特点是可以在某个点暂停执行，并在之后恢复执行，而不会阻塞整个线程。这种特性使得协程非常适合处理异步任务、事件驱动编程以及协作式多任务处理。通过使用协程，可以编写更简洁、易于理解和维护的异步代码。协程可以避免回调地狱，使得代码逻辑更加清晰，并且能够有效地管理共享资源和处理并发任务。

协程又称用户空间线程，调度权归属于应用程序，操作系统无法对协程进行调度。协程的开销小，支持数量众多；因此，让百万协程千万吞吐量的服务成为了可能。

在 Swoole 中的应用

Swoole 中协程的调度器，是利用 EventLoop 事件循环来实现的。例如：在执行某个协程的过程中遇到了类似 MySQL->query() 的 IO 操作，则会把这个 MySQL 连接放到事件循环中。然后让出 CPU 给其他的协程，此时的协程处于挂起状态。直到 MySQL 返回数据，才会恢复当前协程继续执行后续代码。此外，协程在单进程模式下无法利用多核的 CPU，要利用多核CPU需要开启多进程模式。

Swoole 中的事件循环，换而言之其底层逻辑使用的操作系统的IO多路复用。IO多路复用技术的本质就是使用一个进程来维护多个 Socket，这个 Socket 连接可以是网络 Socket、文件 Socket，与之相对应的是网络IO、磁盘IO。其中用的最多的是 epoll，底层采用红黑树的数据结构来实现。红黑树是个高效的数据结构，增删改一般时间复杂度是 O(logn)。

<?php

use Swoole\Coroutine;

function task() {
    echo "协程开始\n";
    Coroutine::sleep(1); // 模拟耗时操作
    echo "协程结束\n";
}

// 创建多个协程
Coroutine::create('task');
Coroutine::create('task');
Coroutine::create('task');

// 让当前进程等待所有协程完成
Coroutine::wait();

echo "所有协程都已完成\n";

在 Go 语言中的应用

Go 语言中协程调度的是使用 GMP 模型实现的，G-M-P 分别代表：G - Goroutine，Go 协程，是参与调度与执行的最小单位；M - Machine，指的是系统级线程；P - Processor，指的是逻辑处理器，P 关联了的本地可运行 G 的队列(也称为LRQ)，最多可存放256个G。

GMP调度流程大致如下：

线程 M 要运行任务就需得获取 P，即与 P 关联。
从 P 的本地队列 (LRQ) 获取 G。
LRQ 中没有可运行的G，M会尝试从全局队列 (GRQ) 拿一批G放到P的本地队列。
全局队列未找到可运行的 G，那么 M 会从其他 P 的本地队列拿一半放到自己 P 的本地队列。
拿到可运行的 G 之后，M 运行 G，G 执行之后，M 会从 P 获取下一个 G，不断重复下去。

package main

import (
    "fmt"
    "sync"
)

func task(workerID int, wg *sync.WaitGroup) {
    defer wg.Done()

    fmt.Printf("协程 %d 开始\n", workerID)
    // 执行其他业务逻辑

    fmt.Printf("协程 %d 结束\n", workerID)
}

func main() {
    workerCount := 4
    var wg sync.WaitGroup

    for i := 0; i < workerCount; i++ {
        wg.Add(1)
        go task(i, &wg)
    }

    wg.Wait()

    fmt.Println("所有协程都已完成")
}