- #block_server.py 非阻塞 IO 示例
- # 有个疑惑: 下面的 connfd 的 blockind 要设置为 True, 不然会出错, 待解决
- from socket import *
- from time import ctime,sleep
- s = socket()
- s.bind(("0.0.0.0",18888))
- s.listen(10)
- # 设置 s 为非阻塞状态
- s.setblocking(False)
- while True:
- print("等待连接...")
- try:
- connfd,addr = s.accept()
- except BlockingIOError:
- sleep(2)
- print(ctime())
- continue
- print("connect from",addr)
- while True:
- connfd.setblocking(True)
- data = connfd.recv(1024)
- if not data:
- break
- print(data.decode())
- serv_msg = input("服务端消息>")
- connfd.send(serv_msg.encode())
- connfd.close()
- s.close()
- =========================================
- #timeout_server.py 超时检测示例及 traceback
- from socket import *
- from time import sleep,ctime
- import traceback
- HOST = "0.0.0.0"
- PORT = 14444
- s = socket()
- s.bind((HOST,PORT))
- s.listen(5)
- # 设置超时检测
- s.settimeout(5)
- while True:
- print("服务端等待连接.....")
- try:
- c,addr = s.accept()
- except Exception:
- traceback.print_exc()
- continue
- print("connect from",addr)
- while True:
- data = c.recv(1024)
- if not data:
- break
- print(data.decode())
- serv_msg = input("服务端消息>")
- c.send(serv_msg.encode())
- c.close()
- s.close()
- ==========================================
- #select_server.py select 服务端示例
- from socket import *
- from select import *
- HOST = "0.0.0.0"
- PORT = 18888
- s = socket()
- s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
- s.bind((HOST,PORT))
- s.listen(5)
- # 将需要监控的 IO 放入 rlist
- rlist = [s]
- wlist = []
- xlist = []
- print("select 服务器启动成功")
- while True:
- print(f"select 正在监控阻塞套接字{rlist}")
- print(f"目前列表中有 {len(rlist)} 个套接字")
- rl,wl,xl = select(rlist,wlist,xlist)
- for r in rl:
- #套接字准备就绪
- if r is s:
- conn,addr = r.accept()
- print("Connect from",addr)
- #把新的套接字加入监控列表
- rlist.append(conn)
- else:
- data = r.recv(1024)
- if not data:
- rlist.remove(r)
- r.close()
- break
- print("received:",data.decode())
- r.send("已经收到".encode())
- for w in wl:
- pass
- for x in xl:
- pass
- ===============================================
- #select_server2_wlist.py select 服务端示例, 演示 wlist,xlist 用法
- from socket import *
- from select import *
- import sys
- HOST = "0.0.0.0"
- PORT = 18888
- s = socket()
- s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
- s.bind((HOST,PORT))
- s.listen(5)
- # 将需要监控的 IO 放入 rlist
- rlist = [s]
- wlist = []
- xlist = [s]
- print("select 服务器启动成功")
- while True:
- print(f"select 正在监控阻塞套接字{rlist}")
- print(f"目前列表中有 {len(rlist)} 个套接字")
- rl,wl,xl = select(rlist,wlist,xlist)
- for r in rl:
- #套接字准备就绪
- if r is s:
- conn,addr = r.accept()
- print("Connect from",addr)
- #把新的套接字加入监控列表
- rlist.append(conn)
- else:
- data = r.recv(1024)
- if not data:
- rlist.remove(r)
- r.close()
- break
- else:
- print(f"received from {str(r.getpeername())}:{data.decode()}",)
- #接收完消息, 把套接字加入 wlist 列表, 待处理(当 wlist 内有内容, select 会立即返回)
- wlist.append(r)
- # 当上一步待处理套接字加入 wlist, 这步立刻处理, 处理完把该套接字移出 wlist
- for w in wl:
- w.send(f"已经收到你的消息{str(w.getpeername())}".encode())
- wlist.remove(w)
- for x in xl:
- #如果套接字出错, 结束整个程序
- if x is s:
- s.close()
- sys.exit()
- =============================================================
- #unix_send.py 本地套接字简单应用, 发送端
- from socket import *
- import sys,os
- #确定用哪个文件进行通信
- server_address = "./test"
- #判断文件是否存在, 如已存在则删除
- if os.path.exists(server_address):
- os.unlink(server_address)
- #创建本地套接字
- s = socket(AF_UNIX,SOCK_STREAM)
- #绑定本地套接字文件
- s.bind(server_address)
- #监听
- s.listen(5)
- #收发消息
- while True:
- c,addr = s.accept()
- while True:
- data = c.recv(1024)
- if data:
- print(data.decode())
- c.sendall("收到消息".encode())
- else:
- break
- c.close()
- s.close()
- ========================================================
- #unix_recv.py 本地套接字简单应用, 接收端
- from socket import *
- server_address = "./test"
- s = socket(AF_UNIX,SOCK_STREAM)
- s.connect(server_address)
- while True:
- message = input("客户端消息")
- if message:
- s.sendall(message.encode())
- recv_msg = s.recv(1024)
- print(recv_msg.decode())
- else:
- break
- s.close()
来源: http://www.bubuko.com/infodetail-3424392.html