前言
前面我们大概简单理解了一下并发编程涉及的一些概念，也简单举例说明相关并发并行和GLI这些知识点。
接下来我们在开始进入asyncio中的协程之前继续探讨一下并发编程涉及到一些概念，阻塞和非阻塞。
阻塞（Blocking）和非阻塞（Non-blocking）
阻塞和非阻塞是计算机科学中描述程序执行中线程（或进程）状态的概念。这些概念用于描述线程或进程在执行过程中的行为，尤其是在它们需要等待某些操作完成时
阻塞的种类有多种（多数的情况下，我们一般只谈所谓的阻塞IO多一些）。 在操作系统中，阻塞可以细分为几种类型：
进程阻塞：当一个进程在执行过程中需要等待某个资源（如CPU、内存、I/O设备等）时，它会被阻塞。例如，一个进程可能因为等待I/O操作完成而被阻塞。
线程阻塞：在多线程环境中，一个线程在等待另一个线程释放锁或完成某个操作时，可能会被阻塞。
网络阻塞：在网络编程中，当一个进程或线程在等待网络通信的响应时，它可能会被阻塞。例如，一个客户端在发送请求后等待服务器的响应时，可能会阻塞。
所以我们可以总结为：阻塞操作是一般是指当一个线程或进程在执行I/O操作（如网络请求或文件读写）的时候，如果该操作尚未完成或无法立即返回结果，那么我们当前线程或进程会被挂起，直到I/O操作完成或准备好数据后再不会阻塞。在这个阻塞的这个过程中，线程或进程不能执行其他任务。
与阻塞相反，非阻塞操作是指在进行I/O操作时，如果操作不能立即完成，线程或进程不会挂起等待，而是继续执行其他任务。在非阻塞模式下，I/O操作通常会立即返回一个状态码，指示操作是否成功，或者是否需要稍后重试。
换个方式来说，就是：
阻塞就像你在餐厅排队等待点餐。在你前面的顾客正在点餐，而你只能耐心等待，不能去干别的事情。当前面的顾客点完餐，服务员才会轮到你，这时你才能进行点餐。在这个过程中，你被“阻塞”了，无法进行其他活动。
而我们的非阻塞就像是你在餐厅里，不需要排队等待，可以直接点餐。你告诉服务员你的需求，然后就可以去干别的事情，比如找个座位坐下，或者看看菜单。当你的餐点准备好了，服务员会叫你过去取餐，而不会影响你正在做的事情。
由上的非阻塞的说明来看，我们大概可以知道非阻塞的主要好处是在于提高了资源的使用效率和程序的响应性。在非阻塞模式下，线程或进程在等待某些操作完成的时候还可以继续执行其他的任务，这样可以充分的利用了我们的CPU资源，也提高了程序的吞吐量以及我们的响应速度。
下面举一个简单的套接字的示例来说明一下阻塞，首先我们定义一个服务端，如下代码所示：
import socket
# 创建一个阻塞的套接字
# socket.AF_INET 表示IPv4地址家族，而 socket.SOCK_STREAM 指定套接字类型为流（用于TCP连接）
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# ('127.0.0.1', 8888) 是一个元组，其中 127.0.0.1 是本地地址（即回环地址，任何发到此地址的包都会被送回本地机器），8888 是端口号。这个服务器将在这个地址和端口上监听
server.bind(('127.0.0.1', 8888))
# listen 方法使得服务器能够接受连接。参数 5 指定了一个最大连接数，即服务器同时可以处理的客户端连接数。
server.listen(5)
print("服务器正在监听端口 8888...")
# 无限循环，等待客户端连接
while True:
try:
# accept() 方法会阻塞，直到有客户端连接到服务器。一旦有客户端连接，它将返回一个新的套接字对象 conn 和地址元组 addr。
conn, addr = server.accept()
print(f"连接来自：{addr}")
# recv 方法接收来自客户端的数据，参数 1024 指定了一次可以接收的最大数据量（以字节为单位）
data = conn.recv(1024)
print(f"接收到数据：{data.decode()}")
# 创建了一个响应字符串，将其编码为字节串，并加上从客户端接收到的数据，然后使用 sendall 方法将整个响应发送回客户端。
response = "这是服务器的响应".encode('utf-8')+data
conn.sendall(response)
# 关闭连接
conn.close()
except Exception as e:
print(f"发生错误：{e}")
上面的代码中我们创建了一个基于TCP协议的Python服务器。该服务器监听本地IP地址为127.0.0.1的8888端口，服务器会不断地监听端口8888，等待客户端的连接请求。当有客户端连接时，服务器就会接收客户端发送的数据，然后接收客户端发送的数据，发送响应数据给客户端，然后关闭连接。如果在执行过程中发生异常，服务器会打印异常信息。
需注意，我们目前开启的这个服务代码是一个简单的示例，它并没有使用多线程或多进程进行客户端请求链接的处理，因此它一次只能处理一个客户端的连接。通常在实际的生产环境中，我们可能是需要使用多线程或多进程来同时处理多个客户端的连接滴哦。
此时我们先启动服务器会得到如下的结果：
服务器正在监听端口 8888...
有了服务端猴，接下里我们开始编写一个客户端来发送数据给服务端，如下代码所示：
import socket
# 创建一个阻塞的套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置超时时间，例如设置为5秒
client_socket.settimeout(5)
try:
# 尝试连接到服务器，这个操作可能会阻塞，直到连接成功或超时
client_socket.connect(('127.0.0.1', 8888))
except socket.timeout:
print("连接超时")
except Exception as e:
print("连接失败：", e)
# 发送数据
try:
client_socket.sendall(b'你好！我的是你大爷!')
except Exception as e:
print("发送数据失败：", e)
# 接收数据，这个操作可能会阻塞，因此会一直等道接收到数据或超时
try:
data = client_socket.recv(1024)
if data:
print("接收到数据：", data.decode())
except socket.timeout:
print("接收数据超时")
except Exception as e:
print("接收数据失败：", e)
# 关闭套接字
client_socket.close()
上面的客户端代码代码执行流程如下：
首先创建套接字并设置超时。 再接着尝试连接到服务器，如果成功则继续，超时或失败则打印相关信息。 如果连接成功，发送数据到服务器。 接收来自服务器的数据，并打印。 完成上述操作后，关闭套接字。
在上面的示例中，我们服务端同一个时间只能处理一个客户端的连接，为了处理更多的连接，我们可以使用多线程的方式，如下一个简单的示例代码所示：
import socket
import threading
# 创建一个阻塞的套接字
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('127.0.0.1', 8888)) # 绑定IP地址和端口号
server.listen(5) # 开始监听连接数限制
print("服务器正在监听端口 8888...")
# 一个函数，用于处理客户端连接
def handle_client(conn, addr):
print(f"连接来自：{addr}") # 打印连接地址
try:
while True:
# 接收客户端发送的数据
data = conn.recv(1024) # 接收数据，最大长度为1024字节
if not data:
break
print(f"接收到数据：{data.decode()}") # 打印接收到的数据
# 模拟服务器处理业务逻辑耗时
time.sleep(2)
# 发送响应数据给客户端
response = "这是服务器的响应：我告诉了你!哦".encode('utf-8') + data # 构造响应数据
conn.sendall(response) # 发送数据
except Exception as e:
print(f"发生错误：{e}")
finally:
# 关闭连接
conn.close()
# 无限循环，等待客户端连接
while True:
try:
# 接受客户端连接
conn, addr = server.accept()
# 创建一个新的线程来处理这个连接
client_thread = threading.Thread(target=handle_client, args=(conn, addr))
client_thread.start()
except Exception as e:
print(f"发生错误：{e}")
和前端的对比，我们发现我们在无限循环，等待客户端连接客户端时候为每个链接的都单独的开启了一个线程去处理，这样就达到了处理多个链接的效果。当然这个只是一个简单的示例，如果合适共享资源的之类的话，上面的代码还是不完善的哦！
有了额服务端的代码，接下来我们开始编写一个并发压测的验证示例，如下代码所示：
import socket
import threading
# 定义要发送的数据
MESSAGE = b'Hello, Server!'
# 定义发送数据的函数
def send_data(client_socket):
try:
client_socket.sendall(MESSAGE)
except Exception as e:
print("发送数据失败：", e)
# 定义接收数据的函数
def receive_data(client_socket):
try:
data = client_socket.recv(1024)
if data:
print("接收到数据：", data.decode())
except Exception as e:
print("接收数据失败：", e)
# 定义连接服务器的函数
def connect_server():
# 创建一个阻塞的套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置超时时间，例如设置为5秒
client_socket.settimeout(5)
try:
# 尝试连接到服务器，这个操作可能会阻塞，直到连接成功或超时
client_socket.connect(('127.0.0.1', 8888))
print("连接成功")
# 发送和接收数据
send_data(client_socket)
receive_data(client_socket)
except socket.timeout:
print("连接超时")
except Exception as e:
print("连接失败：", e)
finally:
# 关闭套接字
client_socket.close()
# 定义并发测试函数
def run_concurrent_test(num_threads):
# 创建线程列表
threads = []
# 启动指定数量的线程
for i in range(num_threads):
thread = threading.Thread(target=connect_server)
thread.start()
threads.append(thread)
# 等待所有线程完成
for thread in threads:
thread.join()
# 运行并发测试
if __name__ == "__main__":
num_threads = 5 # 指定线程数量
run_concurrent_test(num_threads)
接下来运行我们的服务端和客户端，观察服务端输出的结果如下：
服务器正在监听端口 8888...
连接来自：('127.0.0.1', 57597)
接收到数据：我是你大爷
连接来自：('127.0.0.1', 57598)
接收到数据：我是你大爷
连接来自：('127.0.0.1', 57599)
接收到数据：我是你大爷连接来自：('127.0.0.1', 57600)
接收到数据：我是你大爷
连接来自：('127.0.0.1', 57601)
接收到数据：我是你大爷
客户端输出：
连接成功连接成功
连接成功
连接成功
连接成功
接收到数据： 这是服务器的响应：我告诉了你!哦我是你大爷
接收到数据： 这是服务器的响应：我告诉了你!哦我是你大爷
接收到数据：接收到数据：接收到数据： 这是服务器的响应：我告诉了你!哦我是你大爷
这是服务器的响应：我告诉了你!哦我是你大爷这是服务器的响应：我告诉了你!哦我是你大爷
上面我们实现了相关的多线程方式来避免非阻塞的示例说明。
需注意：
# 创建一个阻塞的套接字
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
我们这个地方创建是一个阻塞的套接字哦！
接下里是广告插播，上面关于阻塞和阻塞的，在我的个人新书《FastAPI Web开发入门、进阶与实战》
也有提到，具体的内容如下：
感兴趣的话可以，京东购买哦！
非阻塞套接字
在前面我们为了实现‘非’阻塞服务端，开启了多线程的方式处理每个客户端的请求处理。
每个连接的客户端可以有一个单独的线程来处理其请求，而主线程可以继续监听新的连接请求。 在多线程环境中，主线程可以创建一个线程池来管理多个客户端连接。当一个客户端连接时，主线程会创建一个新的线程来处理该连接，而不会阻塞主线程的执行。 这种方式可以实现并发处理多个客户端连接，同时保持每个连接的响应性。
它的主要优势是：
易于实现：多线程的阻塞套接字可以使用现成的线程池库，如concurrent.futures，使得并发编程更加容易。 性能：阻塞套接字在处理大量数据时通常比非阻塞套接字更高效，因为它们可以一次性处理大量数据，而不是分批处理。 并发处理：多线程可以有效地处理多个客户端连接，使得服务器可以同时处理多个请求
但是这种处理方式的存在的缺点主要是：
资源消耗：每个线程都会消耗一定的系统资源，如果线程数量过多，可能会导致资源耗尽。
线程间同步：多线程程序需要考虑线程间的同步问题，以避免竞态条件和数据不一致。
阻塞主线程：如果主线程在等待子线程完成时没有正确地处理其他任务，可能会导致主线程阻塞。
上面这种并发处理模式可能比较合适于非实时性要求的，可以容忍一定延迟且需要处理大量数据的应用。对于其他实时性要求高的，我们可以基于非阻塞性套接字来实现，如下代码所示；
import socket
import select
# 创建套接字并监听端口
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 设置socket选项，允许地址复用
server_socket.bind(('localhost', 8888)) # 绑定到本地的8888端口
server_socket.listen(5) # 开始监听，最大连接数为5
# 将服务套接字加入到select对象中，作为监视对象
inputs = [server_socket]
while True:
# 使用select模块进行轮询
readables, _, _ = select.select(inputs, [], [])
for readable in readables:
if readable == server_socket:
# 如果是服务套接字可读，则有新的连接请求
client_socket, client_address = server_socket.accept()
print(f"新连接来自 {client_address}")
# 将新连接的客户端套接字加入到select对象中，作为监视对象
inputs.append(client_socket)
else:
# 如果是已连接的客户端套接字可读，则有数据到达
data = readable.recv(1024)
if data:
# 处理收到的数据
print(f"收到来自 {readable.getpeername()} 的数据：{data.decode('utf-8')}")
# 向客户端发送回显数据
readable.sendall(data)
else:
# 客户端关闭了连接，将其从select对象中移除
print(f"{readable.getpeername()} 的连接已关闭")
inputs.remove(readable)
readable.close()
在上面的代码中，我们使用Python的socket模块和select模块实现了一个简单TCP服务器。代码从开始执行就创建一个服务套接字并监听指定端口，然后使用select模块进行轮询，当有新的连接请求或数据到达时进行相应处理。当有新连接时，接受连接并打印客户端地址，当接收到数据时打印来源地址并发送回显数据。当客户端连接关闭时，从select对象中移除该客户端套接字并关闭连接。
此时我们继续使用前面的客户端代码进行请求测试(使用另一种方式写，但是注意着地方其实并发没什么用哦！)：
import asyncio
import socket
# 服务器的IP地址和端口号
SERVER_IP = '127.0.0.1' # 服务器IP地址
SERVER_PORT = 8888 # 服务器端口号
# 定义要发送的消息列表
messages = ["我是你大爷！!", "我是你二舅", "我是你二舅爷"] # 要发送的消息列表
# 创建一个异步函数来发送数据
async def send_data(message):
# 创建异步socket客户端
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # 设置保活属性
client_socket.ioctl(socket.SIO_KEEPALIVE_VALS, (1, 60 * 1000, 30 * 1000)) # 设置保活时间
client_socket.connect((SERVER_IP, SERVER_PORT)) # 连接服务器
try:
# 发送数据给服务器
client_socket.sendall(message.encode('utf-8')) # 发送数据
print(f"Sent: {message}") # 打印发送的消息
# 接收数据，这个操作可能会阻塞，直到接收到数据或超时
try:
data = client_socket.recv(1024) # 接收数据
if data:
print("接收到数据：", data.decode()) # 打印接收到的数据
except socket.timeout:
print("接收数据超时") # 超时未收到数据
except Exception as e:
print("接收数据失败：", e) # 接收数据失败
except Exception as e:
print(f"Error sending message: {e}") # 发送数据失败
finally:
# 关闭连接
client_socket.close() # 关闭socket连接
# 使用asyncio创建协程并运行
async def main():
# 创建任务列表
tasks = []
for message in messages:
tasks.append(asyncio.create_task(send_data(message)))
# 并发执行任务
await asyncio.gather(*tasks)
# 启动事件循环
if __name__ == "__main__":
asyncio.run(main()) # 运行事件循环
运行客户端后，观察客户端输出结果为：
Sent: 我是你大爷！!
接收到数据： 我是你大爷！!
Sent: 我是你二舅
接收到数据： 我是你二舅
Sent: 我是你二舅爷
接收到数据： 我是你二舅爷
至此我们也编写了一个简单的非阻塞型的TCP服务。
文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！
结尾
END
简书：https://www.jianshu.com/u/d6960089b087
掘金：https://juejin.cn/user/2963939079225608
公众号：微信搜【程序员小钟同学】
新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107
小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822
原文作者：程序员小钟同学；如有侵权，请联系管理员删除。
最新文章
- 深入FastAPI异步编程-之并发编程（15）-定义全局的支持全局上下文变量和局部上下文的对象可以任意传递了！
- 深入FastAPI异步编程-之并发编程（13）-纠错-修改基于aiojobs库结合aiojobs的和FastAPI框架的示例
- 深入FastAPI异步编程-之并发编程（13）-基于aiojobs库实现类似Background Tasks的效果
- 深入FastAPI异步编程-之并发编程（12）-基于pyee异步事件驱动通知
- 深入FastAPI异步编程-之并发编程（11）-异步中的?同步原语
- 深入FastAPI异步编程-之并发编程（10）-?asyncio.wait 更精细化并发补充说明
- 深入FastAPI异步编程-之并发编程（9）-asyncio.as_completed的补充说明
- 深入FastAPI异步编程-之并发编程（8）-asyncio.gather的补充说明
- 深入FastAPI异步编程-之并发编程（7）番外篇~Windows系统下事件循环插曲
- 深入FastAPI异步编程-之并发编程（6）~事件循环
