绕过 GIL 的雷区：并发 Python 的冒险指南

import multiprocessingdef task(n):# 执行 CPU 密集型任务return n * nif __name__ == "__main__":pool = multiprocessing.Pool(4)# 创建一个进程池results = pool.map(task, range(10000))# 使用进程池执行任务print(results)

import concurrent.futuresdef task(n):# 执行 CPU 密集型任务return n * nif __name__ == "__main__":with concurrent.futures.ThreadPoolExecutor() as executor:results = executor.map(task, range(10000))# 使用 GIL 友好的线程池执行任务print(results)

#include <python.h>static PyObject* task(PyObject* self, PyObject* args) {int n;if (!PyArg_ParseTuple(args, "i", &amp;n)) {return NULL;}// 执行 CPU 密集型任务int result = n * n;return Py_BuildValue("i", result);}static PyMethodDef methods[] = {{"task", task, METH_VARARGS, "Task function"},{NULL, NULL, 0, NULL}};static struct PyModuleDef module = {PyModuleDef_HEAD_INIT,"mymodule",NULL,-1,methods};PyMODINIT_FUNC PyInit_mymodule(void) {return PyModule_Create(&amp;module);}</python.h>

import asyncioasync def task(n):# 执行 CPU 密集型任务return n * nasync def main():tasks = [task(i) for i in range(10000)]results = await asyncio.gather(*tasks)# 并行执行任务print(results)if __name__ == "__main__":asyncio.run(main())