如何在 asyncio 中使用 socket

简介: 如何在 asyncio 中使用 socket


楔子



本次我们来聊一聊 Socket,以及它如何与 asyncio 搭配使用。


阻塞 Socket



Socket 是对 TCP/IP 协议的一个封装,可以让我们更方便地使用 TCP/IP 协议,而不用关注背后的原理。并且我们经常使用的 Web 框架,本质上也是一个 Socket。

所以 Socket 是操作系统对 TCP/IP 网络协议栈的封装,并提供了一系列的接口,我们通过这些接口可以实现网络通信,而不用关注网络协议的具体细节。

按照现有的网络模型,Socket 并不属于其中的任何一层,但我们可以简单地将 Socket 理解为传输层之上的抽象层,负责连接应用层和传输层。Socket 提供了大量的 API,基于这些 API 我们可以非常方便地使用网络协议栈,在不同主机间进行网络通信。

Linux 一切皆文件,Socket 也不例外,它被称为套接字文件,在使用上和普通文件是类似的。

Socket 是什么我们已经知道了,下面来看看如何使用 Socket 进行编程。

869439742439b5574daf6c5613e906d8.png

整个过程如下:

  • 服务端初始化 socket,此时会得到「主动套接字」;
  • 服务端调用 bind 方法,将套接字绑定在某个 IP 和端口上;
  • 服务端调用 listen 进行监听,此时「主动套接字」会变成「监听套接字」;
  • 服务端调用 accept,等待客户端连接,此时服务端会阻塞在这里(调用的是阻塞的 API);
  • 客户端同样初始化 socket,得到主动套接字;
  • 客户端调用主动套接字的 connect,向服务器端发起连接请求,如果连接成功,后续客户端就用这个主动套接字进行数据的传输;
  • 当客户端来连接时,那么服务端的 accept 将不再阻塞,并返回「已连接套接字」,后续服务端便用这个已连接套接字和客户端进行数据传输;
  • 当客户端来连接时,那么服务端的 accept 将不再阻塞,并返回「已连接套接字」,后续服务端便用这个已连接套接字和客户端进行数据传输;

我们使用来编写代码演示一下这个过程,首先是服务端:

import socket
# socket.socket() 会返回一个「主动套接字」
server = socket.socket(
    # 表示使用 IPv4,如果是 socket.AF_INET6
    # 则表示使用 IPv6
    socket.AF_INET,
    # 表示建立 TCP 连接,如果是 socket.SOCK_DGRAM
    # 则表示建立 UDP 连接
    socket.SOCK_STREAM
)
# 当然这两个参数也可以不传,因为默认就是它
# 设置套接字属性,这里让端口释放后立刻就能再次使用
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
# 将「主动套接字」绑定在某个 IP 和端口上
server.bind(("localhost", 12345))
# 监听,此时「主动套接字」会变成「监听套接字」
server.listen(5)
# 调用 accept,等待客户端连接,此时会阻塞在这里
# 如果客户端连接到来,那么会返回「已连接套接字」,也就是这里的 conn
# 至于 addr 则是一个元组,保存了客户端连接的信息(IP 和端口)
conn, addr = server.accept()
# 下面我们通过「已连接套接字」conn 和客户端进行消息的收发
# 收消息使用 recv、发消息使用 send,和 read、write 本质是一样的
while True:
    msg = conn.recv(1024)
    # 当客户端断开连接时,msg 会收到一个空字节串
    if not msg:
        print("客户端已经断开连接")
        conn.close()
        break
    print("客户端发来消息:", msg.decode("utf-8"))
    # 然后我们加点内容之后,再给客户端发过去
    conn.send("服务端收到, 你发的消息是: ".encode("utf-8") + msg)

接下来编写客户端:

import socket
# 返回主动套接字
client = socket.socket(socket.AF_INET,
                       socket.SOCK_STREAM)
# 连接服务端
client.connect(("localhost", 12345))
while True:
    # 发送消息
    data = input("请输入内容: ")
    if data.strip().lower() in ("q", "quit", "exit"):
        client.close()
        print("Bye~~~")
        break
    client.send(data.encode("utf-8"))
    print(client.recv(1024).decode("utf-8"))

启动服务端和客户端进行测试:

4d1471bf3e447600a1f13ffbca6bd03c.png

还是比较简单的,当然我们这里的服务端每次只能和一个客户端通信,如果想服务多个客户端的话,那么需要为已连接套接字单独开一个线程和客户端进行通信,然后主线程继续调用 accept 方法等待下一个客户端。

下面来编写一下多线程的版本,这里只需要编写服务端即可,客户端代码不变。

import socket
import threading
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
server.bind(("localhost", 12345))
server.listen(5)
def handle_message(conn, addr):
    while True:
        msg = conn.recv(1024)
        if not msg:
            print(f"客户端(ip: {addr[0]}, port: {addr[1]}) 已经断开连接")
            conn.close()
            break
        print(f"客户端(ip: {addr[0]}, port: {addr[1]}) 发来消息:",
              msg.decode("utf-8"))
        conn.send("服务端收到, 你发的消息是: ".encode("utf-8") + msg)
while True:
    conn, addr = server.accept()
    threading.Thread(
        target=handle_message,
        args=(conn, addr)
    ).start()

代码很简单,就是把已连接套接字和客户端的通信逻辑写在了单独的函数中,每来一个客户端,服务端都会启动一个新的线程去执行该函数,然后继续监听,等待下一个客户端连接到来。

然后客户端代码不变,我们启动三个客户端去和服务端通信,看看结果如何。

f954c6c9ddde2206ed922a059dc808a7.png

结果一切正常,当然我们这里的代码比较简单,就是普通的消息收发。你也可以实现一个更复杂的功能,比如文件下载器,把服务端当成网盘,支持客户端上传和下载文件,并不难。


非阻塞 Socket



先回顾一下 socket 模型:

a44b917e54d89563f5fa2ecf090d94e7.png

但是注意:我们说在 listen() 这一步,会将主动套接字转化为监听套接字,但此时的监听套接字的类型是阻塞的。阻塞类型的监听套接字在调用 accept() 方法时,如果没有客户端来连接的话,就会一直处于阻塞状态,那么此时主线程就没法干其它事情了。

所以要设置为非阻塞,而非阻塞的监听套接字在调用 accept() 时,如果没有客户端来连接,那么主线程不会傻傻地等待,而是会直接返回,然后去做其它的事情。

类似的,我们在创建已连接套接字的时候默认也是阻塞的,阻塞类型的已连接套接字在调用 send() 和 recv() 的时候也会处于阻塞状态。比如当客户端一直不发数据的时候,已连接套接字就会一直阻塞在 recv() 这一步。如果是非阻塞类型的已连接套接字,那么当调用 recv() 但却收不到数据时,也不用处于阻塞状态,同样可以直接返回去做其它事情。

import socket
server = socket.socket()
server.bind(("localhost", 12345))
# 调用 setblocking 方法,传入 False
# 表示将监听套接字和已连接套接字的类型设置为非阻塞
server.setblocking(False)
server.listen(5)
while True:
    try:
        # 非阻塞的监听套接字调用 accept() 时
        # 如果发现没有客户端连接,则会立刻抛出 BlockingIOError
        # 因此这里写了个死循环
        conn, addr = server.accept()
    except BlockingIOError:
        pass
    else:
        break
while True:
    try:
        # 同理,非阻塞的已连接套接字在调用 recv() 时
        # 如果发现客户端没有发数据,那么同样会报错
        msg = conn.recv(1024)
    except BlockingIOError:
        pass
    else:
        print(msg.decode("utf-8"))
        conn.send(b"data from server")

很明显,虽然上面的代码在运行的时候正常,但存在两个问题:

1)虽然 accept() 不阻塞了,在没有客户端连接时主线程可以去做其它事情,但如果后续有客户端连接,主线程要如何得知呢?因此必须要有一种机制,能够继续在监听套接字上等待后续连接请求,并在请求到来时通知主线程。我们上面的做法是写了一个死循环,但很明显这是没有意义的,这种做法还不如使用阻塞的套接字。

2)send() / recv() 不阻塞了,相当于 I/O 读写流程不再是阻塞的,读写方法都会瞬间完成并返回,也就是说它会采用能读多少就读多少、能写多少就写多少的策略来执行 I/O 操作,这显然更符合我们对性能的追求。

c5269b21b67451be20e493a937e62776.png

显然对于非阻塞套接字而言,会面临一个问题,那就是当我们执行读取操作时,有可能只读了一部分数据,剩余的数据客户端还没发过来,那么这些数据何时可读呢?同理写数据也是这种情况,当缓冲区满了,而我们的数据还没有写完,那么剩下的数据又何时可写呢?因此同样要有一种机制,能够在主线程做别的事情的时候继续监听已连接套接字,并且在有数据可读写的时候通知主线程。

这样才能保证主线程既不会像基本 IO 模型一样,一直在阻塞点等待,也不会无法处理实际到达的客户端连接请求和可读写的数据,而上面所提到的机制便是 I/O 多路复用。

早期的所有框架都是非阻塞 + 回调 + 基于 IO 多路复用的事件循环,这种模式的性能也非常高,Redis 和 Nginx 都是基于这种方式实现了高并发。只是这种编码方式非常痛苦,它将好端端的自上而下的逻辑分割的四分五裂,而且也不好维护,它使得开发人员在编写业务逻辑的同时,还要关注并发细节。

因此使用多路复用 + 回调的方式编写异步化代码,虽然并发量能上去,但是对开发者很不友好;而使用同步的方式编写同步代码,虽然很容易理解,可并发量却又上不去。那么问题来了,有没有一种办法,能够让我们在享受异步化带来的高并发的同时,又能以同步的方式去编写代码呢?也就是我们能不能以同步的方式去编写异步化的代码呢?

答案是可以的,使用「协程」便可以办到。协程在这种模式的基础之上又批了一层外衣,兼顾了开发效率与运行效率。


在 asyncio 中使用 Socket



asyncio 的事件循环提供了处理套接字的一些方法,我们主要会用到三个:

  • sock_accept()
  • sock_recv()
  • sock_sendall()

这些方法类似于前面使用的套接字方法,但不同之处在于,它们需要接收非阻塞套接字作为参数,然后返回协程。我们可以等待协程,直到有数据可供操作。

先来看一下 sock_accept(),它类似于 server.accept()。

conn,add = await loop.sock_accept(sock)

然后 sock_recv 和 sock_sendall 的调用方式与 sock_accept 类似,它们接收一个套接字,然后返回协程对象。通过 await 表达式,sock_recv 将会阻塞,直到套接字有可以处理的字节;sock_sendall 接收一个套接字和要发送的数据,同样会陷入阻塞,直到要发送给套接字的所有数据都发送完毕,成功时返回 None。

data = await loop.sock_recv(sock)
await loop.sock_sendall(sock, data)

下面我们就基于 asyncio 设计一个回显服务器。

import asyncio
import socket
async def echo(conn: socket.socket):
    loop = asyncio.get_running_loop()
    # 无限循环等待来自客户端连接的数据
    try:
        while data := await loop.sock_recv(conn, 1024):
            # 收到数据之后再将其发送给客户端
            # 为了区分,我们发送的时候在结尾加一个 b"~"
            await loop.sock_sendall(conn, data + b"~")
    except Exception as e:
        print(f"服务出错: {e}")
    finally:
        conn.close()
async def listen_for_conn(server: socket.socket):
    loop = asyncio.get_running_loop()
    while True:
        conn, addr = await loop.sock_accept(server)
        conn.setblocking(False)
        print(f"收到客户端 {addr} 的连接")
        # 每次连接时,都创建一个任务来监听客户端的数据
        asyncio.create_task(echo(conn))
async def main():
    server = socket.socket()
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
    server.setblocking(False)
    server.bind(("localhost", 12345))
    server.listen()
    await listen_for_conn(server)
asyncio.run(main())

运行这个应用程序可以同时服务多个客户端,它里面同样使用了 IO 多路复用,只不过事件循环将它封装起来了,我们不需要直接面对。所以这种编程模式就简单多了。


小结



如果使用阻塞套接字创建应用程序,那么阻塞套接字将在等待数据时停止整个线程。这阻止了我们实现并发,因为一次只能从一个客户端获取数据。

使用非阻塞套接字构建应用程序,这些套接字总是会立即返回,而结果有两种:要么已经准备好了数据,要么因为没有数据而出现异常。

使用 asyncio 的事件循环方法来构建具有非阻塞套接字的应用程序,这些方法接收一个套接字并返回一个协程,然后可在 await 表达式中使用它。这将暂停父协程,直到套接字带有数据。事件循环就是基于 IO 多路复用做的一个封装,而 IO 多路复用能够实现的前提之一就是:套接字必须是非阻塞的。

相关文章
|
达摩院 供应链
「达摩院MindOpt」用于多目标规划(加权和法)
多目标规划(Multi-objective programming)是指在一个优化问题中需要同时考虑多个目标函数的优化。在多目标规划问题中,目标函数之间通常是互相冲突的,即在优化一个目标函数的过程中,另一个或几个目标函数可能会受到影响。因此,多目标规划问题的目标是找到一个解x,使得在满足约束的前提下,所有目标函数达到一个相对满意的折中。
「达摩院MindOpt」用于多目标规划(加权和法)
|
JavaScript
vue elementUI select下拉框设置默认值
vue elementUI select下拉框设置默认值
2358 0
|
8月前
|
数据采集 监控 网络协议
基于aiohttp的高并发爬虫实战:从原理到代码的完整指南
在数据驱动时代,传统同步爬虫效率低下,而基于Python的aiohttp库可构建高并发异步爬虫。本文通过实战案例解析aiohttp的核心组件与优化策略,包括信号量控制、连接池复用、异常处理等,并探讨代理集成、分布式架构及反爬应对方案,助你打造高性能、稳定可靠的网络爬虫系统。
635 0
Python 代码从 `.env` 文件中读取环境变量
这篇文章介绍了如何在Python项目中使用`python-dotenv`库从`.env`文件读取环境变量的详细步骤,包括安装库、创建`.env`文件、在代码中加载和读取环境变量。
|
JavaScript 网络协议 前端开发
【Nodejs】WebSocket 全面解析+实战演练——(Nodejs实现简易聊天室)
【Nodejs】WebSocket 全面解析+实战演练——(Nodejs实现简易聊天室)
1072 0
|
资源调度 前端开发 数据可视化
构建高效的数据可视化仪表板:D3.js与React的融合之道
【10月更文挑战第25天】在数据驱动的时代,将复杂的数据集转换为直观、互动式的可视化表示已成为一项至关重要的技能。本文深入探讨了如何结合D3.js的强大可视化功能和React框架的响应式特性来构建高效、动态的数据可视化仪表板。文章首先介绍了D3.js和React的基础知识,然后通过一个实际的项目案例,详细阐述了如何将两者结合使用,并提供了实用的代码示例。无论你是数据科学家、前端开发者还是可视化爱好者,这篇文章都将为你提供宝贵的洞见和实用技能。
465 5
|
前端开发 Java API
解密 asyncio 的 Future 和 Task
解密 asyncio 的 Future 和 Task
688 2
|
关系型数据库 BI Python
Python异步编程|PySimpleGUI界面读取PDF转换Excel
实例要求:使用PySimpleGUI库做一个把单位考勤系统导出的pdf文件合并输出Excel表格的应用,库pdfplumber直接遍历多个pdf文件,得到数据后输出xks文件,同时对pdf文件读取函数的进行改造,使用了asyncio异步编程效果非常不错。
327 2
|
数据采集 网络协议 Python
requests和aiohttp中代理IP的使用
requests和aiohttp中代理IP的使用
1044 3
|
存储 Kubernetes 安全
kubernetes集群备份与恢复
k8s集群服务所有组件都是无状态服务,所有数据都存储在etcd集群当中,所以为保证k8s集群的安全可以直接备份etcd集群数据,备份etcd的数据相当于直接备份k8s整个集群。
2742 1