将socket通信变成并发的方式
时间:2022-04-23
本文章向大家介绍将socket通信变成并发的方式，主要内容包括一 利用multiprocessing模块，开启多进程，实现socket通信并发、2.多进程实现socket并发通信、3 进程池实现并发通信、二 利用threading模块，开启多线程，实现socket通信并发、2.多进程实现socket并发通信、三 利用concurrent模块，开启进程池、线程池实现socket通信并发、2. 线程池、四 利用gevent模块，协程实现单线程下的socket通信并发、五 利用selectors模块，实现socket并发通信、基本概念、基础应用、原理机制和需要注意的事项等，并结合实例形式分析了其使用技巧，希望通过本文能帮助到大家理解应用这部分内容。
一 利用multiprocessing模块，开启多进程，实现socket通信并发
1. 开启子进程的两种方式
import time
import random
from multiprocessing import Process
def piao(name):
print('%s piaoing' %name)
time.sleep(random.randrange(1,5))
print('%s piao end' %name)
p1=Process(target=piao,args=('egon',)) #必须加,号
p2=Process(target=piao,args=('alex',))
p3=Process(target=piao,args=('wupeqi',))
p4=Process(target=piao,args=('yuanhao',))
p1.start()
p2.start()
p3.start()
p4.start()
print('主线程')#开进程的方法二:
import time
import random
from multiprocessing import Process
class Piao(Process): #注意一定要继承Process
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
print('%s piaoing' %self.name)
time.sleep(random.randrange(1,5))
print('%s piao end' %self.name)
p1=Piao('egon')
p2=Piao('alex')
p3=Piao('wupeiqi')
p4=Piao('yuanhao')
p1.start() #start会自动调用run
p2.start()
p3.start()
p4.start()
print('主进程')2.多进程实现socket并发通信
服务端
from socket import *
from multiprocessing import Process
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8081))
server.listen(5)
def talk(conn,client_addr):
while True:
try: #若不用此句，客户端关闭时，服务端会因报错，停止
msg=conn.recv(1024)
if not msg:break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__': #windows下start进程一定要写到这下面
while True:
conn,client_addr=server.accept()
p=Process(target=talk,args=(conn,client_addr))
p.start()客户端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8081))
while True:
msg=input('>>: ').strip()
if not msg:
continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))存在的问题：每来一个客户端，都在服务端开启一个进程，如果并发来一个万个客户端，要开启一万个进程吗，你自己尝试着在你自己的机器上开启一万个，10万个进程试一试。
解决办法：进程池3 进程池实现并发通信
3 进程池实现并发通信
使用进程池维护固定数目的进程
#Pool内的进程数默认是cpu核数，假设为4（查看方法os.cpu_count()）
#开启6个客户端，会发现2个客户端处于等待状态
#在每个进程内查看pid，会发现pid使用为4个，即多个客户端公用4个进程
from socket import *
from multiprocessing import Pool
import os
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)
def talk(conn,client_addr):
print('进程pid: %s' %os.getpid())
while True:
try:
msg=conn.recv(1024)
if not msg:break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
p=Pool()
while True:
conn,client_addr=server.accept()
p.apply_async(talk,args=(conn,client_addr))
# p.apply(talk,args=(conn,client_addr)) #同步的话，则同一时间只有一个客户端能访问from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8081))
while True:
msg=input('>>: ').strip()
if not msg:
continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))发现：并发开启多个客户端，服务端同一时间只有4个不同的pid，干掉一个客户端，另外一个客户端才会进来，被3个进程之一处理
二 利用threading模块，开启多线程，实现socket通信并发
1. 开启多线程的两种方式
#方式一
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('egon',))
t.start()
print('主线程')#方式二
from threading import Thread
import time
class Sayhi(Thread): #注意继承Thread类
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
time.sleep(2)
print('%s say hello' % self.name)
if __name__ == '__main__':
t = Sayhi('egon')
t.start()
print('主线程')2.多进程实现socket并发通信
服务端
from threading import Thread
from socket import *
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8081))
server.listen(5)
def talk(conn,addr):
while True:
try:
msg=conn.recv(1024)
if not msg:
break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
while True:
conn,addr=server.accept()
t=Thread(target=talk,args=(conn,addr))
t.start()客户端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8081))
while True:
msg=input('>>: ').strip()
if not msg:
continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))三 利用concurrent模块，开启进程池、线程池实现socket通信并发
1. 进程池
服务端
from concurrent.futures import ProcessPoolExecutor
from socket import *
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8081))
server.listen(5)
def talk(conn,addr):
while True:
try:
msg=conn.recv(1024)
if not msg:
break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
p=ProcessPoolExecutor() #不填则默认为cpu的个数
while True:
conn,addr=server.accept()
p.submit(talk,conn,addr)客户端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8081))
while True:
msg=input('>>: ').strip()
if not msg:
continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))2. 线程池
服务端
from concurrent.futures import ThreadPoolExecutor
from socket import *
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8081))
server.listen(5)
def talk(conn,addr):
while True:
try:
msg=conn.recv(1024)
if not msg:
break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
p=ThreadPoolExecutor() #不填则默认为cpu的个数*5
while True:
conn,addr=server.accept()
p.submit(talk,conn,addr)客户端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8081))
while True:
msg=input('>>: ').strip()
if not msg:
continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))四 利用gevent模块，协程实现单线程下的socket通信并发
通过gevent实现单线程下的socket并发（from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前，否则gevent无法识别socket的阻塞）
from gevent import monkey;monkey.patch_all()
from socket import *
import gevent
#如果不想用money.patch_all()打补丁,可以用gevent自带的socket
# from gevent import socket
# s=socket.socket()
def server(server_ip,port):
s=socket(AF_INET,SOCK_STREAM)
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind((server_ip,port))
s.listen(5)
while True:
conn,addr=s.accept()
gevent.spawn(talk,conn,addr)
def talk(conn,addr):
try:
while True:
res=conn.recv(1024)
print('client %s:%s msg: %s' %(addr[0],addr[1],res))
conn.send(res.upper())
except Exception as e:
print(e)
finally:
conn.close()
if __name__ == '__main__':
server('127.0.0.1',8080)from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
msg=input('>>: ').strip()
if not msg:continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))from threading import Thread
from socket import *
import threading
def client(server_ip,port):
c=socket(AF_INET,SOCK_STREAM) #套接字对象一定要加到函数内，即局部名称空间内，放在函数外则被所有线程共享，则大家公用一个套接字对象，那么客户端端口永远一样了
c.connect((server_ip,port))
count=0
while True:
c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
msg=c.recv(1024)
print(msg.decode('utf-8'))
count+=1
if __name__ == '__main__':
for i in range(500):
t=Thread(target=client,args=('127.0.0.1',8080))
t.start()五 利用selectors模块，实现socket并发通信
selectors模块，帮我们默认选择当前平台下最合适的IO多路复用模型（select、poll和epoll）
#服务端
from socket import *
import selectors
sel=selectors.DefaultSelector()
def accept(server_fileobj,mask):
conn,addr=server_fileobj.accept()
sel.register(conn,selectors.EVENT_READ,read)
def read(conn,mask):
try:
data=conn.recv(1024)
if not data:
print('closing',conn)
sel.unregister(conn)
conn.close()
return
conn.send(data.upper()+b'_SB')
except Exception:
print('closing', conn)
sel.unregister(conn)
conn.close()
server_fileobj=socket(AF_INET,SOCK_STREAM)
server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server_fileobj.bind(('127.0.0.1',8088))
server_fileobj.listen(5)
server_fileobj.setblocking(False) #设置socket的接口为非阻塞
sel.register(server_fileobj,selectors.EVENT_READ,accept) #相当于网select的读列表里append了一个文件句柄server_fileobj,并且绑定了一个回调函数accept
while True:
events=sel.select() #检测所有的fileobj，是否有完成wait data的
for sel_obj,mask in events:
callback=sel_obj.data #callback=accpet
callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8088))
while True:
msg=input('>>: ')
if not msg:continue
c.send(msg.encode('utf-8'))
data=c.recv(1024)
print(data.decode('utf-8'))
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- Android延时操作的三种方法
- 5步教你快速写一个android Router路由框架
- Android 中使用RecyclerView实现底部翻页
- Android中Glide库的使用小技巧总结
- Android Studio手动配置Gradle的方法
- Android仿微信@好友功能 输入@跳转、删除整块
- Android开发实现广告无限循环功能示例
- Android仿微信底部菜单栏效果
- MySQL 案例：Update 死锁详解
- Android 线程之自定义带消息循环Looper的实例
- 详解Androidstudio3.0 关于Gradle报错的问题(小结)
- Android开发实现文件关联方法介绍
- Android开发获取重力加速度和磁场强度的方法
- Android自定义View圆形和拖动圆跟随手指拖动
- 简单好用的Adapter—ArrayAdapter详解