python中socket、进程、线程、协程、池的创建方式和应用场景
分类:
IT文章
•
2023-11-22 19:33:49
![]()
进程
场景 利用多核、高计算型的程序、启动数量有限
进程是计算机中最小的资源分配单位
进程和线程是包含关系 每个进程中都至少有一条线程
可以利用多核，数据隔离
创建 销毁 切换 时间开销都比较大
随着开启的数量增加 给操作系统带来负担
线程 高IO型 调度是我们不能干预的 我们只能写我们自己的逻辑
场景 一些协程现有的模块不能完成帮助我们规避IO操作的功能 适合使用多线程 urllib
被CPU调度的最小单位，线程的切换时操作系统完成的
在cpython解释器下不能利用多核，数据共享
创建 销毁 切换 时间开销都比进程小很多
随着开启的数量增加 给操作系统带来负担
协程 高IO型 用户可以自己控制的 我们能否抢占更多的资源完全取决于我们切换策略
场景 一些通用的场景 可以用协程现有的模块来规避一些IO操作适合使用协程
协程的切换工作是用户完成的
是一个线程，完全不能利用多核，不会产生数据不安全的现象
多个任务之间互相切换不依赖操作系统，无论开启多少个协程都不会给操作系统带来负担
应用场景
一、TCP-socket
服务端：
import socket
tcp_sk = socket.socket()
tcp_sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
tcp_sk.bind(('127.0.0.1',8000))
tcp_sk.listen()
conn,addr = tcp_sk.accept()
conn.send('你好'.encode('utf-8'))
print(conn.recv(1024).decode('utf-8'))
conn.close()
tcp_sk.close()
客户端：
import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8000))
print(sk.recv(1024).decode('utf-8'))
sk.send('嘿嘿嘿'.encode('utf-8'))
sk.close()
二、UDP-socket
服务端：
import socket
udp_sk = socket.socket(type=socket.SOCK_DGRAM)
udp_sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
udp_sk.bind(('127.0.0.1',8001))
msg,addr = udp_sk.recvfrom(1024)
print(msg.decode('utf-8'))
udp_sk.sendto('你好'.encode('utf-8'),addr)
udp_sk.close()
客户端：
import socket
sk = socket.socket(type=socket.SOCK_DGRAM)
sk.sendto('哈哈'.encode('utf-8'),('127.0.0.1',8001))
msg,addr = sk.recvfrom(1024)
print(msg.decode('utf-8'))
sk.close()
三、socketserver
服务端：
import socketserver
class Myserver(socketserver.BaseRequestHandler):
def handle(self):
conn = self.request
while True:
conn.send(b'hello')
print(conn.recv(1024).decode('utf-8'))
socketserver.TCPServer.allow_reuse_address = True
server = socketserver.ThreadingTCPServer(('127.0.0.1',8080),Myserver)
server.serve_forever()
客户端：
import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8080))
while True:
ret = sk.recv(1024)
print(ret.decode('utf-8'))
sk.send(b'hiworld')
sk.close()
四、进程
方式一、
from multiprocessing import Process
def func(arg):
print(arg)
if __name__ == '__main__':
p = Process(target=func,args=('子进程',))
p.start()
p.join()
print('主进程')
方式二、
from multiprocessing import Process
class MyProcess(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
print(self.name)
if __name__ == '__main__':
p = MyProcess('小明')
p.start()
五、线程
方式一、
from threading import Thread
import time
def sleep_boy(name):
time.sleep(1)
print('%s is sleeping' %name)
t = Thread(target=sleep_boy,args=('xiaoming',)) # 这里可以不需要main，因为现在只是在一个进程内操作，不需要导入进程就不会import主进程了
t.start()
print('主线程')
方式二、
from threading import Thread
import time
class Sleep_boy(Thread):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
time.sleep(1)
print('%s is sleeping' % self.name)
t = Sleep_boy('xiaoming')
t.start()
print('主线程')
六、协程
1、greenlet例子：
import time
from greenlet import greenlet
def cooking():
print('cooking 1')
g2.switch() # 切换到g2，让g2的函数工作
time.sleep(1)
print('cooking 2')
def watch():
print('watch TV 1')
time.sleep(1)
print('watch TV 2')
g1.switch() # 切换到g1，让g1的函数工作
g1 = greenlet(cooking)
g2 = greenlet(watch)
g1.switch() # 切换到g1，让g1的函数工作
greenlet的缺陷：很显然greenlet实现了协程的切换功能，可以自己设置什么时候切，在哪切，但是它遇到阻塞并没有自动切换，
因此并不能提高效率。所以一般我们都使用gevent模块实现协程
2、gevent例子1：
from gevent import monkey
monkey.patch_all()
import time
import gevent
def cooking():
print('cooking 1')
time.sleep(1)
print('cooking 2')
def watch():
print('watch TV 1')
time.sleep(1)
print('watch TV 2')
g1 = gevent.spawn(cooking) # 自动检测阻塞事件，遇见阻塞了就会进行切换
g2 = gevent.spawn(watch)
g1.join() # 阻塞直到g1结束
g2.join() # 阻塞直到g2结束
gevent例子2：
import gevent
def cooking(i):
print('%s号在煮饭' %i)
return i
g_lst = []
for i in range(10):
g = gevent.spawn(cooking,i) # 函数名，参数
g_lst.append(g) # 把协程对象放入列表
for g in g_lst:
g.join()
print(g.value) # 打印返回值
# gevent.joinall(g_lst) # joinall一次性把全部对象都阻塞
七、进程池
1、同步提交apply：
import os
import time
from multiprocessing import Pool
def test(num):
time.sleep(1)
print('%s:%s' %(num,os.getpid()))
return num*2
if __name__ == '__main__':
p = Pool()
for i in range(20):
res = p.apply(test,args=(i,)) # 提交任务的方法 同步提交
print('-->',res) # res就是test的return的值，同步提交的返回值可以直接使用
2、异步提交apply_async：
2-1无返回值：
import time
from multiprocessing import Pool
def func(num):
time.sleep(1)
print('做了%s件衣服'%num)
if __name__ == '__main__':
p = Pool(4) # 进程池中创建4个进程，不写的话，默认值为你电脑的CUP数量
for i in range(50):
p.apply_async(func,args=(i,)) # 异步提交func到一个子进程中执行，没有返回值的情况
p.close() # 关闭进程池，用户不能再向这个池中提交任务了
p.join() # 阻塞，直到进程池中所有的任务都被执行完
2-2有返回值：
import time
import os
from multiprocessing import Pool
def test(num):
time.sleep(1)
print('%s:%s' %(num,os.getpid()))
return num*2
if __name__ == '__main__':
p = Pool()
res_lst = []
for i in range(20):
res = p.apply_async(test,args=(i,)) # 提交任务的方法 异步提交
res_lst.append(res)
for res in res_lst:
print(res.get()) # 异步提交的返回值需要get，get有阻塞效果，此时就不需要close和join
2-3map:
map接收一个函数和一个可迭代对象，是异步提交的简化版本，自带close和join方法
可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的任务数量
map可以直接拿到返回值的可迭代对象(列表)，循环就可以获取返回值
import time
from multiprocessing import Pool
def func(num):
print('子进程：',num)
# time.sleep(1)
return num
if __name__ == '__main__':
p = Pool()
ret = p.map(func,range(10)) # ret是列表
for i in ret:
print('返回值：',i)
2-4回调函数：
import os
from multiprocessing import Pool
def func(i):
print('子进程：',os.getpid())
return i
def call_back(res):
print('回调函数：',os.getpid())
print('res--->',res)
if __name__ == '__main__':
p = Pool()
print('主进程：',os.getpid())
p.apply_async(func,args=(1,),callback=call_back) # callback关键字传参，参数是回调函数
p.close()
p.join()
八、进程池、线程池
线程池：
1、
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('thread',i)
time.sleep(1)
print('thread %s end'%i)
tp = ThreadPoolExecutor(5) # 相当于tp = Pool(5)
tp.submit(func,1) # 相当于tp.apply_async(func,args=(1,))
tp.shutdown() # 相当于tp.close() + tp.join()
print('主线程')
2、
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
print('thread',i,currentThread().ident)
time.sleep(1)
print('thread %s end'%i)
tp = ThreadPoolExecutor(5)
for i in range(20):
tp.submit(func,i)
tp.shutdown() # shutdown一次就够了，会自动把所有的线程都join()
print('主线程')
3、返回值
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
print('thread',i,currentThread().ident)
time.sleep(1)
print('thread %s end' %i)
return i * '*'
tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
ret = tp.submit(func,i)
ret_lst.append(ret)
for ret in ret_lst:
print(ret.result()) # 相当于ret.get()
print('主线程')
4、map
map接收一个函数和一个可迭代对象
可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的线程数量
map可以直接拿到返回值的可迭代对象(列表)，循环就可以获取返回值
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('thread',i)
time.sleep(1)
print('thread %s end'%i)
return i * '*'
tp = ThreadPoolExecutor(5)
ret = tp.map(func,range(20))
for i in ret:
print(i)
5、回调函数
回调函数在进程池是由主进程实现的
回调函数在线程池是由子线程实现的
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
print('thread',i,currentThread().ident)
time.sleep(1)
print('thread %s end'%i)
return i * '*'
def call_back(arg):
print('call back : ',currentThread().ident)
print('ret : ',arg.result()) # multiprocessing的Pool回调函数中的参数不需要get(),这里需要result()
tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
tp.submit(func,i).add_done_callback(call_back) # 使用add_done_callback()方法实现回调函数
print('主线程',currentThread().ident)
![]()
进程
场景 利用多核、高计算型的程序、启动数量有限
进程是计算机中最小的资源分配单位
进程和线程是包含关系 每个进程中都至少有一条线程
可以利用多核，数据隔离
创建 销毁 切换 时间开销都比较大
随着开启的数量增加 给操作系统带来负担
线程 高IO型 调度是我们不能干预的 我们只能写我们自己的逻辑
场景 一些协程现有的模块不能完成帮助我们规避IO操作的功能 适合使用多线程 urllib
被CPU调度的最小单位，线程的切换时操作系统完成的
在cpython解释器下不能利用多核，数据共享
创建 销毁 切换 时间开销都比进程小很多
随着开启的数量增加 给操作系统带来负担
协程 高IO型 用户可以自己控制的 我们能否抢占更多的资源完全取决于我们切换策略
场景 一些通用的场景 可以用协程现有的模块来规避一些IO操作适合使用协程
协程的切换工作是用户完成的
是一个线程，完全不能利用多核，不会产生数据不安全的现象
多个任务之间互相切换不依赖操作系统，无论开启多少个协程都不会给操作系统带来负担
应用场景
一、TCP-socket
服务端：
import socket
tcp_sk = socket.socket()
tcp_sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
tcp_sk.bind(('127.0.0.1',8000))
tcp_sk.listen()
conn,addr = tcp_sk.accept()
conn.send('你好'.encode('utf-8'))
print(conn.recv(1024).decode('utf-8'))
conn.close()
tcp_sk.close()
客户端：
import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8000))
print(sk.recv(1024).decode('utf-8'))
sk.send('嘿嘿嘿'.encode('utf-8'))
sk.close()
二、UDP-socket
服务端：
import socket
udp_sk = socket.socket(type=socket.SOCK_DGRAM)
udp_sk.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
udp_sk.bind(('127.0.0.1',8001))
msg,addr = udp_sk.recvfrom(1024)
print(msg.decode('utf-8'))
udp_sk.sendto('你好'.encode('utf-8'),addr)
udp_sk.close()
客户端：
import socket
sk = socket.socket(type=socket.SOCK_DGRAM)
sk.sendto('哈哈'.encode('utf-8'),('127.0.0.1',8001))
msg,addr = sk.recvfrom(1024)
print(msg.decode('utf-8'))
sk.close()
三、socketserver
服务端：
import socketserver
class Myserver(socketserver.BaseRequestHandler):
def handle(self):
conn = self.request
while True:
conn.send(b'hello')
print(conn.recv(1024).decode('utf-8'))
socketserver.TCPServer.allow_reuse_address = True
server = socketserver.ThreadingTCPServer(('127.0.0.1',8080),Myserver)
server.serve_forever()
客户端：
import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8080))
while True:
ret = sk.recv(1024)
print(ret.decode('utf-8'))
sk.send(b'hiworld')
sk.close()
四、进程
方式一、
from multiprocessing import Process
def func(arg):
print(arg)
if __name__ == '__main__':
p = Process(target=func,args=('子进程',))
p.start()
p.join()
print('主进程')
方式二、
from multiprocessing import Process
class MyProcess(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
print(self.name)
if __name__ == '__main__':
p = MyProcess('小明')
p.start()
五、线程
方式一、
from threading import Thread
import time
def sleep_boy(name):
time.sleep(1)
print('%s is sleeping' %name)
t = Thread(target=sleep_boy,args=('xiaoming',)) # 这里可以不需要main，因为现在只是在一个进程内操作，不需要导入进程就不会import主进程了
t.start()
print('主线程')
方式二、
from threading import Thread
import time
class Sleep_boy(Thread):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
time.sleep(1)
print('%s is sleeping' % self.name)
t = Sleep_boy('xiaoming')
t.start()
print('主线程')
六、协程
1、greenlet例子：
import time
from greenlet import greenlet
def cooking():
print('cooking 1')
g2.switch() # 切换到g2，让g2的函数工作
time.sleep(1)
print('cooking 2')
def watch():
print('watch TV 1')
time.sleep(1)
print('watch TV 2')
g1.switch() # 切换到g1，让g1的函数工作
g1 = greenlet(cooking)
g2 = greenlet(watch)
g1.switch() # 切换到g1，让g1的函数工作
greenlet的缺陷：很显然greenlet实现了协程的切换功能，可以自己设置什么时候切，在哪切，但是它遇到阻塞并没有自动切换，
因此并不能提高效率。所以一般我们都使用gevent模块实现协程
2、gevent例子1：
from gevent import monkey
monkey.patch_all()
import time
import gevent
def cooking():
print('cooking 1')
time.sleep(1)
print('cooking 2')
def watch():
print('watch TV 1')
time.sleep(1)
print('watch TV 2')
g1 = gevent.spawn(cooking) # 自动检测阻塞事件，遇见阻塞了就会进行切换
g2 = gevent.spawn(watch)
g1.join() # 阻塞直到g1结束
g2.join() # 阻塞直到g2结束
gevent例子2：
import gevent
def cooking(i):
print('%s号在煮饭' %i)
return i
g_lst = []
for i in range(10):
g = gevent.spawn(cooking,i) # 函数名，参数
g_lst.append(g) # 把协程对象放入列表
for g in g_lst:
g.join()
print(g.value) # 打印返回值
# gevent.joinall(g_lst) # joinall一次性把全部对象都阻塞
七、进程池
1、同步提交apply：
import os
import time
from multiprocessing import Pool
def test(num):
time.sleep(1)
print('%s:%s' %(num,os.getpid()))
return num*2
if __name__ == '__main__':
p = Pool()
for i in range(20):
res = p.apply(test,args=(i,)) # 提交任务的方法 同步提交
print('-->',res) # res就是test的return的值，同步提交的返回值可以直接使用
2、异步提交apply_async：
2-1无返回值：
import time
from multiprocessing import Pool
def func(num):
time.sleep(1)
print('做了%s件衣服'%num)
if __name__ == '__main__':
p = Pool(4) # 进程池中创建4个进程，不写的话，默认值为你电脑的CUP数量
for i in range(50):
p.apply_async(func,args=(i,)) # 异步提交func到一个子进程中执行，没有返回值的情况
p.close() # 关闭进程池，用户不能再向这个池中提交任务了
p.join() # 阻塞，直到进程池中所有的任务都被执行完
2-2有返回值：
import time
import os
from multiprocessing import Pool
def test(num):
time.sleep(1)
print('%s:%s' %(num,os.getpid()))
return num*2
if __name__ == '__main__':
p = Pool()
res_lst = []
for i in range(20):
res = p.apply_async(test,args=(i,)) # 提交任务的方法 异步提交
res_lst.append(res)
for res in res_lst:
print(res.get()) # 异步提交的返回值需要get，get有阻塞效果，此时就不需要close和join
2-3map:
map接收一个函数和一个可迭代对象，是异步提交的简化版本，自带close和join方法
可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的任务数量
map可以直接拿到返回值的可迭代对象(列表)，循环就可以获取返回值
import time
from multiprocessing import Pool
def func(num):
print('子进程：',num)
# time.sleep(1)
return num
if __name__ == '__main__':
p = Pool()
ret = p.map(func,range(10)) # ret是列表
for i in ret:
print('返回值：',i)
2-4回调函数：
import os
from multiprocessing import Pool
def func(i):
print('子进程：',os.getpid())
return i
def call_back(res):
print('回调函数：',os.getpid())
print('res--->',res)
if __name__ == '__main__':
p = Pool()
print('主进程：',os.getpid())
p.apply_async(func,args=(1,),callback=call_back) # callback关键字传参，参数是回调函数
p.close()
p.join()
八、进程池、线程池
线程池：
1、
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('thread',i)
time.sleep(1)
print('thread %s end'%i)
tp = ThreadPoolExecutor(5) # 相当于tp = Pool(5)
tp.submit(func,1) # 相当于tp.apply_async(func,args=(1,))
tp.shutdown() # 相当于tp.close() + tp.join()
print('主线程')
2、
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
print('thread',i,currentThread().ident)
time.sleep(1)
print('thread %s end'%i)
tp = ThreadPoolExecutor(5)
for i in range(20):
tp.submit(func,i)
tp.shutdown() # shutdown一次就够了，会自动把所有的线程都join()
print('主线程')
3、返回值
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
print('thread',i,currentThread().ident)
time.sleep(1)
print('thread %s end' %i)
return i * '*'
tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
ret = tp.submit(func,i)
ret_lst.append(ret)
for ret in ret_lst:
print(ret.result()) # 相当于ret.get()
print('主线程')
4、map
map接收一个函数和一个可迭代对象
可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的线程数量
map可以直接拿到返回值的可迭代对象(列表)，循环就可以获取返回值
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
print('thread',i)
time.sleep(1)
print('thread %s end'%i)
return i * '*'
tp = ThreadPoolExecutor(5)
ret = tp.map(func,range(20))
for i in ret:
print(i)
5、回调函数
回调函数在进程池是由主进程实现的
回调函数在线程池是由子线程实现的
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
print('thread',i,currentThread().ident)
time.sleep(1)
print('thread %s end'%i)
return i * '*'
def call_back(arg):
print('call back : ',currentThread().ident)
print('ret : ',arg.result()) # multiprocessing的Pool回调函数中的参数不需要get(),这里需要result()
tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
tp.submit(func,i).add_done_callback(call_back) # 使用add_done_callback()方法实现回调函数
print('主线程',currentThread().ident)
