python_0">python全栈-并发和网络通信

• 并发
• 线程thread
• 进程Process
• 协程/纤程coroutine

进程>线程>协程

常用的生成模式：

• 多线程
• 多进程
• 多线程+多进程

线程

使用threading库的Thread模块

• 参数：
  • target后面写方法名
  • args给方法的参数，元组的形式
• start 使用这个启动线程

创建线程的方式

1. 直接使用threading模块，去执行对应的函数

# coding='utf-8'
import threading
import timedef addd(name):for i in range(3):print(f"addd{name}--{i}")time.sleep(1) threading.Thread(target=addd, args=(1,)).start()
threading.Thread(target=addd, args=(2,)).start()

2. 使用类方法去继承thread，然后重定义run方法

# coding='utf-8'
import time
from threading import Threadclass mythread(Thread):def __init__(self, name):Thread.__init__(self)self.name = namedef run(self):for i in range(3):print(self.name,i)time.sleep(1)mythread('th1').start()
mythread('th2').start()

等待线程join

由于某种原因，我们在创建子线程的时候，不希望子线程可以脱离主线程自己跑

就是想要把子线程包在主线程里面

需要使用join方法

# coding='utf-8'
import time
from threading import Threadclass mythread(Thread):def __init__(self, name):Thread.__init__(self)self.name = namedef run(self):for i in range(3):print(self.name,i)time.sleep(1)if __name__ == '__main__':print('1')th1 = mythread('th1')th2 = mythread('th2')th1.start()th2.start()th1.join()th2.join()print('2')

就像start方法一样，把线程使用join方法就ok了，主线程就会一直等待子线程执行完所有程序再执行后面的

守护线程daemon=true/false

如果我们在多线程模式下，既有子线程又有主线程，在不使用join的方法时，主线程可能会比子线程跑的更快。我们希望在主线程结束的时候，子线程也能结束，就需要使用守护线程daemon

# coding='utf-8'
import time
from threading import Threadclass mythread(Thread):def __init__(self, name):Thread.__init__(self)self.name = namedef run(self):for i in range(3):print(self.name,i)time.sleep(1)if __name__ == '__main__':print('1')th1 = mythread('th1')th1.daemon = True  需要在start之前th1.start()print('2')

当主线程的print（2）输出之后，子线程立即结束

最典型的应用是垃圾收集器，程序结束的时候所有子任务结束

互斥锁lock

• 加锁 lock.acquice()
• 解锁 lock.release()

# coding='utf-8'
import threading
from time import sleepclass person():def __init__(self, name):self.name = nameself.num = 100def consume(number,person):if person.num <= number:print('ERROR')else:sleep(1)person.num -= numberprint(person.num)if __name__ == '__main__':teacher = person('Teacher')threading.Thread(target=consume, args=(80,teacher)).start()threading.Thread(target=consume, args=(80,teacher)).start()

上面是互斥锁的应用场景，发现我们在判断账户可以操作的时候，会有程序阻塞，导致数据无法实时刷新，出现了-60

因此我们需要卡线程的数量，一次只允许一个线程去操作，就有了锁，一次进去一个线程

# coding='utf-8'
import threading
from time import sleepclass person():def __init__(self, name):self.name = nameself.num = 100def consume(number,person):lock.acquire()if person.num <= number:print('ERROR')else:sleep(1)person.num -= numberprint(person.num)lock.release()if __name__ == '__main__':teacher = person('Teacher')lock = threading.Lock()threading.Thread(target=consume, args=(80,teacher)).start()threading.Thread(target=consume, args=(80,teacher)).start()

我们只需要把阻塞的地方锁起来，先加锁，再解锁。就可以保证程序正常运行了

• 死锁：一个程序里面同时含有两个及以上的锁，由于线程的差异，导致线程之间互相等待。只要不同时使用两个及以上的锁就行。

信号量semaphore

就是锁的升级版，因为锁一次只能进一个线程，对于某些场景来说，可以允许两个线程同时操作

就有了信号量的概念

信号量，在初始化的时候，可以定义线程的最大数量，这个信号量就允许同时几个线程运行

和锁的用法一样

当信号量为1的时候，就是普通的锁了

# coding='utf-8'
import threading
from threading import Semaphore
from time import sleepclass person():def __init__(self, name):self.name = nameself.num = 1000def consume(number,person):se.acquire()if person.num <= number:print('ERROR')else:sleep(1)person.num -= numberprint(person.num)se.release()if __name__ == '__main__':teacher = person('Teacher')se = Semaphore(2)  同时进两个线程for i in range(10):threading.Thread(target=consume, args=(80,teacher)).start()

事件event

设置一个对象，专门用来传递信号的，就像过红绿灯一样。所有人到达红绿灯，如果是红灯都要等着。

一旦变成绿灯，就过马路。正常行驶。

• 等红绿灯 event.wait(timeout=time) 可以设置最大等待时间，比如红绿灯损坏，人不能一直等着
• 红绿灯变绿了 event.set()
• 初始化红绿灯 threading.Event()
• 红绿灯变红灯 event.clear() 把有wait的线程阻塞
• 判断当前红绿灯的状态 event.is_set() 看看什么状态

# coding='utf-8'
import threading
import time
from threading import Semaphore
from time import sleepclass person():def __init__(self, name):self.name = nameself.num = 1000def consume(number,person,name):print(name,11)ev.wait()print(name,22)if person.num <= number:print('ERROR')else:sleep(1)person.num -= numberprint(person.num)if __name__ == '__main__':teacher = person('Teacher')ev = threading.Event()for i in range(5):threading.Thread(target=consume, args=(80,teacher,i)).start()time.sleep(10)ev.set() 启动所有线程

协程coroutines

这里的知识，工作用的时候少，面试的时候问的多

除了cpu密集型应用

异步io/asyncio

• async 声明异步函数
• await 声明程序挂起
• gather把异步的函数传入
• run方法去运行主函数

底层逻辑是事件循环

import asyncioasync def fun():for i in range(5):await asyncio.sleep(1)print(f'fun-{i}')print('fun-ok')
async def fun2():for i in range(5):await asyncio.sleep(1)print(f'fun2-{i}')print('fun2-ok')async def main():await asyncio.gather(fun(), fun2()) 添加异步方法if __name__ == '__main__':asyncio.run(main())  执行异步函数

大概分成3部分：

1. 使用run去执行主函数
2. 使用gather去添加异步任务
3. 编写异步方法

计算机网络通信

IP

ip地址，是计算机的身份证

ipv4是旧版的ip地址，32位

ipv6是新版的，128位

本机ip地址是127.0.0.1

在命令行使用ipconfig获取网卡信息

使用ping可以查看网络连接：

• ping + 网址 查看能不能上网
• ping + ip 看看和该计算机在不在一个局域网
• ping + 127.0.0.1 查看本机网卡能不能用

OSI网络协议

网络之间数据传输的层数：

1. 应用层
2. 表示层
3. 会话层
4. 传输层
5. 网络层
6. 数据链路层
7. 物理层

每一层都有自己的协议

TCP/IP协议模型：

1. 应用层 —1~3—socket
2. 传输层 —4 – TCP/UDP
3. 网络层 ----5
4. 物理+数据链路层6~7

TCP/UDP区别

TCP	UDP
一对一	一对多
可靠	不可靠/不稳定
需要连接	不需要连接
类似打电话，需要专门连接	类似广播，不需要连接，能接收信息就行
需要很多的验证信息，20-60字节	几乎不需要验证信息，8字节
文件传输	直播，视频会议

TCP三次握手四次挥手

就是使用TCP进行网络通信的时候，计算机之间需要进行三次通话，才能开始发送真正的数据

在结束TCP的时候，需要进行四次通话，才能完美的结束通话

• 三次握手
  1. 发送端发送SYN数据包，SYN=1，seq=x
  2. 接受端发送SYN/ACK数据包,SYN=1,ACK=1,seq=y,ack=x+1
  3. 发送端发送ACK数据包,ACK=1,seq=x+1,ack=y+1

三次握手结束后，开始发送数据

因为UDP不需要验证，不需要建立连接，没有三次握手

• 四次挥手
  1. 发送端FIN=1，发出TCP停止请求
  2. 接收端ACK=1
  3. 接受端FIN=1，提出关闭请求
  4. 发送端ACK=1，结束

为了断开连接

通信传输的数据单位—数据包

数据包含有—包，帧，数据包，段，消息

socket编程

• socket.socket([family[,type[,proto]]]) 表示打开一个网络连接
• family是套接字家族：AF_UNIX(本地)/AF_INET(ipv4)
• type是套接字类型，TCP的类型是SOCK_STREAM,UDP是SOCK_DGRAM
• proto指定协议不写，默认是0

内置方法和属性

• 服务器server
  • bind 绑定地址
  • lisen TCP监听
  • accept 接收客户端连接
• 客户端client
  • connect 主动初始化TCP连接
  • connect_ex
• 套接方法
  • recv接收TCP，以字符串的形式返回
  • send发送TCP数据
  • sendall发送完整TCP数据
  • recvfrom接收UDP数据
  • sendto发送UDP数据
  • close关闭
  • getpeername
  • getsockname
  • getsockopt

UDP通信

只需要知道IP和端口号，就可以直接发送数据包

对于UDP来说，客户端和服务端基本类似

from socket import *# 初始化socket
sk = socket(AF_INET, SOCK_DGRAM) # UDP
sk.bind(('127.0.0.1', 8080)) # 绑定端口号
print('正在监听')
while True:txt = sk.recvfrom(1024) # 允许单次接收的最大字节数print(txt[0].decode('gbk'),txt[1])if txt[0].decode('gbk')=='quit':breaksk.close()

注意解码方式，我们发送信息的时候需要encode加编码方式gbk，接收的时候就需要decode解码

from socket import *# 初始化socket
sk = socket(AF_INET, SOCK_DGRAM) # UDP
sk.bind(('127.0.0.1', 8888)) # 绑定端口号
while True:data = input('>>>')sk.sendto(data.encode('gbk'), ('127.0.0.1', 8080))  if data == 'quit':break
sk.close()

• 基于UDP的双向通信

import threading
from socket import *def fasong():while True:data = input('>>>')sk.sendto(data.encode('gbk'), ('127.0.0.1', 8888))if data == 'quit':breaksk.close()def jieshou():while True:txt = sk.recvfrom(1024)  # 允许单次接收的最大字节数print(txt[0].decode('gbk'), txt[1])print('>>>',end='')if txt[0].decode('gbk') == 'quit':breaksk.close()if __name__ == '__main__':# 初始化socketsk = socket(AF_INET, SOCK_DGRAM)  # UDPsk.bind(('127.0.0.1', 8080))  # 绑定端口号th1 = threading.Thread(target=fasong)th2 = threading.Thread(target=jieshou)th1.start()th2.start()th1.join()th2.join() 

客户端

import threading
from socket import *def fasong():while True:data = input('>>>')sk.sendto(data.encode('gbk'), ('127.0.0.1', 8080))if data == 'quit':breaksk.close()def jieshou():while True:txt = sk.recvfrom(1024)  # 允许单次接收的最大字节数print(txt[0].decode('gbk'), txt[1])print('>>>',end='')if txt[0].decode('gbk') == 'quit':breaksk.close()if __name__ == '__main__':# 初始化socketsk = socket(AF_INET, SOCK_DGRAM)  # UDPsk.bind(('127.0.0.1', 8888))  # 绑定端口号th1 = threading.Thread(target=fasong)th2 = threading.Thread(target=jieshou)th1.start()th2.start()th1.join()th2.join()

不难发现，基于UDP的双向通信，服务器和客户端一模一样，除了端口号不同，没有区别

TCP通信

• 单次通信TCP

服务器

from socket import *sk = socket(AF_INET, SOCK_STREAM) # TCP连接
sk.bind(('127.0.0.1', 8080)) # 绑定端口号
sk.listen(5) # 最大监听数量，就是同时接收客户端的数量，一般为5就够了
print('等待连接')
client_socket,client_info = sk.accept() # 端口阻塞，等待客户端连接
recv_data = client_socket.recv(1024) # 最大接收字节
print(recv_data.decode('utf-8'))  客户端信息
print(client_info)  客户端地址
# 断开连接
client_socket.close() 先关闭客户端端口
sk.close() 再关闭服务器端口

客户端

from socket import *sk = socket(AF_INET, SOCK_STREAM) # TCP协议
sk.connect(('127.0.0.1', 8080)) # 连接服务器端口，客户端不需要绑定端口
msg = input('>>>')
sk.send(msg.encode('utf-8')) # 向服务器发送信息
sk.close() # 关闭客户端

• TCP多次通信，双向多次通信

客户端

from socket import *sk = socket(AF_INET, SOCK_STREAM)
sk.connect(('127.0.0.1', 8080))
while True:msg = input('>>>')sk.send(msg.encode('utf-8'))if msg == 'exit':breakserver_recv = sk.recv(1024)  接收服务器信息print(server_recv.decode('utf-8'))sk.close()

服务器

from socket import *sk = socket(AF_INET, SOCK_STREAM) # TCP连接
sk.bind(('127.0.0.1', 8080))
sk.listen(5) # 最大监听数量，就是同时接收客户端的数量，一般为5就够了
print('等待连接')
client_socket,client_info = sk.accept()
while True:recv_data = client_socket.recv(1024) # 最大接收字节print(recv_data.decode('utf-8'))if recv_data.decode('utf-8') == 'exit':breakclient_socket.send(f'已接收到信息:{recv_data.decode("utf-8")}'.encode('utf-8'))发送给客户端提示
# 断开连接
client_socket.close()
sk.close()

• 基于TCP实现多次通话

就是把上面单次通话的操作部分，拆开，分成两个函数，然后使用多线程去调用

把初始化，和结束部分放在主函数里面

import threading
from socket import *def fasong():while True:msg = input('>>>')client_socket.send(msg.encode('utf-8'))if msg == 'exit':print('结束通话')break
def jieshou():while True:recv_data = client_socket.recv(1024)  # 最大接收字节print(recv_data.decode('utf-8'))if recv_data.decode('utf-8') == 'exit':print('结束接收')breakif __name__ == '__main__':sk = socket(AF_INET, SOCK_STREAM)  # TCP连接sk.bind(('127.0.0.1', 8080))sk.listen(5)  # 最大监听数量，就是同时接收客户端的数量，一般为5就够了print('等待连接')client_socket, client_info = sk.accept()print('连接成功')t1 = threading.Thread(target=fasong)t2 = threading.Thread(target=jieshou)t1.start()t2.start()t1.join()t2.join()# 断开连接client_socket.close()sk.close()

客户端

import threading
from socket import *def fasong():while True:msg = input('>>>')sk.send(msg.encode('utf-8'))if msg == 'exit':print('结束通话')breakdef jieshou():while True:server_recv = sk.recv(1024)print(server_recv.decode('utf-8'))if server_recv.decode('utf-8') == 'exit':print('结束接收')break
if __name__ == '__main__':sk = socket(AF_INET, SOCK_STREAM)sk.connect(('127.0.0.1', 8080))t1 = threading.Thread(target=fasong)t2 = threading.Thread(target=jieshou)t1.start()t2.start()t1.join()t2.join()sk.close()