Python开发【第八章】：Socket

参数一：地址簇socket.AF_INET IPv4（默认）
socket.AF_INET6 IPv6socket.AF_UNIX 只能够用于单一的Unix系统进程间通信参数二：类型socket.SOCK_STREAM流式socket , for TCP （默认）
socket.SOCK_DGRAM 数据报式socket , for UDPsocket.SOCK_RAW 原始套接字，普通的套接字无法处理ICMP、IGMP等网络报文，而SOCK_RAW可以；其次，SOCK_RAW也可以处理特殊的IPv4报文；此外，利用原始套接字，可以通过IP_HDRINCL套接字选项由用户构造IP头。
socket.SOCK_RDM 是一种可靠的UDP形式，即保证交付数据报但不保证顺序。SOCK_RAM用来提供对原始协议的低级访问，在需要执行某些特殊操作时使用，如发送ICMP报文。SOCK_RAM通常仅限于高级用户或管理员运行的程序使用。
socket.SOCK_SEQPACKET 可靠的连续数据包服务参数三：协议0（默认）与特定的地址家族相关的协议,如果是 0 ，则系统就会根据地址格式和套接类别,自动选择一个合适的协议

# 服务端
import socket
ip_port = ('127.0.0.1',9999)
sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0)
sk.bind(ip_port)while True:
data,(host,port) = sk.recvfrom(1024)
print(data,host,port)
sk.sendto(bytes('ok', encoding='utf-8'), (host,port))#客户端
import socket
ip_port = ('127.0.0.1',9999)sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0)
while True:
inp = input('数据：').strip()
if inp == 'exit':
break
sk.sendto(bytes(inp, encoding='utf-8'),ip_port)
data = sk.recvfrom(1024)
print(data)sk.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#serverimport socketip_port = ("127.0.0.1",8888)
server = socket.socket()
server.bind(ip_port)
server.listen(5)while True:
print("等待连接........")
conn, addr = server.accept()
while True:
data = conn.recv(1024).decode("utf-8")
if data == "exit":
conn.sendall("exit".encode("utf-8"))
break
elif data == "":
conn.sendall("汪星人星球正要进攻地球。。。。".encode("utf-8"))
else:
print(data)
conn.sendall("继续发送电波".encode("utf-8"))
conn.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#clientimport socketip_port = ("127.0.0.1",8888)
client = socket.socket()
client.connect(ip_port)while True:
info = input("->>>")
client.sendall(info.encode("utf-8"))
data = client.recv(1024).decode("utf-8")
if data == "exit":
break
else:
print(data)
client.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#ssh serverimport socket,osip_port = ("127.0.0.1",9999)
server = socket.socket()
server.bind(ip_port)
server.listen(5)while True:
conn,add = server.accept()
while True:
client_data = conn.recv(1024)
recv_data = client_data.decode("utf-8")
if recv_data == "exit":
break
send_data = os.popen(recv_data).read()
if not send_data:
conn.sendall(client_data+"命令不存在".encode("utf-8"))
else:
conn.sendall(send_data.encode("utf-8"))
conn.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#ssh clientimport socketip_port = ("127.0.0.1",9999)
client = socket.socket()
client.connect(ip_port)while True:
info = input("->>").strip()
if not info:
continue
client.sendall(info.encode("utf-8"))
if info == "exit":
break
server_data = client.recv(1024)
print(server_data.decode("utf-8"))client.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#ssh serverimport socket,osip_port = ("127.0.0.1",9999)
server = socket.socket()
server.bind(ip_port)
server.listen(5)while True:
conn,add = server.accept()
while True:
print("开始收")
client_data = conn.recv(1024)
client_data = client_data.decode("utf-8")
if client_data == "exit": #收到exit 退出
break
send_data = os.popen(client_data).read() #执行命令结果，要发送的数据 send_data = send_data.encode("utf-8") #转换为bytes类型 length = str(len(send_data)) #统计发送数据的长度
conn.sendall(length.encode("utf-8")) #长度以bytes类型发送过去 return_value = conn.recv(1024)
return_value = return_value.decode("utf-8") if return_value == "start":
conn.sendall(send_data)
conn.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#ssh clientimport socketip_port = ("127.0.0.1",9999)
client = socket.socket()
client.connect(ip_port)while True:
cmd = input("->>").strip()
if not cmd: #空字符 重新输入
continue
client.sendall(cmd.encode("utf-8")) #要执行的命令发送过去
if cmd == "exit": #如果为exit 退出连接
break length = client.recv(1024) #数据长度
length = length.decode("utf-8")
length = int(length) #长度转换为int client.sendall("start".encode("utf-8")) #发送字节start sum_data = b"" #初始汇总的数据
while length > 0: #循环收数据
server_data = client.recv(1024)
length -=len(server_data)
sum_data +=server_data
print(sum_data.decode("utf-8")) #打印最终的执行数据client.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport socket,os,hashlibip_port = ("127.0.0.1",9999)
server = socket.socket()
server.bind(ip_port)
server.listen(5)while True:
conn,add = server.accept()
while True:
print("开始")
data = conn.recv(1024) cmd,filename = data.decode().split()
if os.path.isfile(filename):
with open(filename,"rb") as file:
file_size = os.stat(filename).st_size
conn.sendall(str(file_size).encode()) #发送文件大小
ack = conn.recv(1024) #等待ack
m = hashlib.md5()
for line in file:
m.update(line)
conn.sendall(line)
conn.sendall(m.hexdigest().encode())
conn.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lianimport socket,hashlibip_port = ("127.0.0.1",9999)
client = socket.socket()
client.connect(ip_port)while True:
cmd = input("->>").strip()
if not cmd:continue
client.sendall(cmd.encode()) #要执行的命令发送过去
if cmd.startswith("get"):
filename = cmd.split()[1]
server_response = client.recv(1024)
file_total_size = int(server_response.decode())
client.sendall("ok".encode())
with open("%s"%filename,"wb") as file:
revice_size = 0
m = hashlib.md5()
while revice_size < file_total_size:
if file_total_size - revice_size > 1024:
size =1024
else:
size = file_total_size - revice_size
data = client.recv(size)
revice_size += len(data)
file.write(data)
m.update(data)
print(file_total_size,revice_size,file_total_size - revice_size) new_file_md5 = m.hexdigest()
server_file_md5 = client.recv(1024).decode()
print("new",new_file_md5)
print("old",server_file_md5)client.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#scoketserverimport socketserver,osclass Myserver(socketserver.BaseRequestHandler):
def handle(self):
while True:
conn = self.request
# conn,add = server.accept()
while True:
print("开始收")
client_data = conn.recv(1024)
client_data = client_data.decode("utf-8")
if client_data == "exit": #收到exit 退出
break
send_data = os.popen(client_data).read() #执行命令结果，要发送的数据
send_data = send_data.encode("utf-8") #转换为bytes类型 length = str(len(send_data)) #统计发送数据的长度
conn.sendall(length.encode("utf-8")) #长度以bytes类型发送过去 return_value = conn.recv(1024)
return_value = return_value.decode("utf-8") if return_value == "start":
if not send_data: # 如果执行结果为空，表示命令不存在
conn.sendall((client_data +"命令不存在").encode("utf-8"))
else:
conn.sendall(send_data)
conn.close()if __name__ == '__main__':
server = socketserver.ThreadingTCPServer(("127.0.0.1",8888),Myserver)
server.serve_forever()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#-Author-Lian#ssh clientimport socketip_port = ("127.0.0.1",8888)
client = socket.socket()
client.connect(ip_port)while True:
cmd = input("->>").strip()
if not cmd: #空字符 重新输入
continue
client.sendall(cmd.encode("utf-8")) #要执行的命令发送过去
if cmd == "exit": #如果为exit 退出连接
break length = client.recv(1024) #数据长度
length = length.decode("utf-8")
length = int(length) #长度转换为int client.sendall("start".encode("utf-8")) #发送字节start sum_data = "" #初始汇总的数据
while length >= 0: #循环收数据
server_data = client.recv(1024)
length -=1024
sum_data +=server_data.decode("utf-8")
print(sum_data) #打印最终的执行数据client.close()

class BaseServer: """Base class for server classes. Methods for the caller: - __init__(server_address, RequestHandlerClass)
- serve_forever(poll_interval=0.5)
- shutdown()
- handle_request() # if you do not use serve_forever()
- fileno() -> int # for select() Methods that may be overridden: - server_bind()
- server_activate()
- get_request() -> request, client_address
- handle_timeout()
- verify_request(request, client_address)
- server_close()
- process_request(request, client_address)
- shutdown_request(request)
- close_request(request)
- handle_error() Methods for derived classes: - finish_request(request, client_address) Class variables that may be overridden by derived classes or
instances: - timeout
- address_family
- socket_type
- allow_reuse_address Instance variables: - RequestHandlerClass
- socket """ timeout = None def __init__(self, server_address, RequestHandlerClass):
"""Constructor. May be extended, do not override."""
self.server_address = server_address
self.RequestHandlerClass = RequestHandlerClass
self.__is_shut_down = threading.Event()
self.__shutdown_request = False def server_activate(self):
"""Called by constructor to activate the server. May be overridden. """
pass def serve_forever(self, poll_interval=0.5):
"""Handle one request at a time until shutdown. Polls for shutdown every poll_interval seconds. Ignores
self.timeout. If you need to do periodic tasks, do them in
another thread.
"""
self.__is_shut_down.clear()
try:
while not self.__shutdown_request:
# XXX: Consider using another file descriptor or
# connecting to the socket to wake this up instead of
# polling. Polling reduces our responsiveness to a
# shutdown request and wastes cpu at all other times.
r, w, e = _eintr_retry(select.select, [self], [], [],
poll_interval)
if self in r:
self._handle_request_noblock()
finally:
self.__shutdown_request = False
self.__is_shut_down.set() def shutdown(self):
"""Stops the serve_forever loop. Blocks until the loop has finished. This must be called while
serve_forever() is running in another thread, or it will
deadlock.
"""
self.__shutdown_request = True
self.__is_shut_down.wait() # The distinction between handling, getting, processing and
# finishing a request is fairly arbitrary. Remember:
#
# - handle_request() is the top-level call. It calls
# select, get_request(), verify_request() and process_request()
# - get_request() is different for stream or datagram sockets
# - process_request() is the place that may fork a new process
# or create a new thread to finish the request
# - finish_request() instantiates the request handler class;
# this constructor will handle the request all by itself def handle_request(self):
"""Handle one request, possibly blocking. Respects self.timeout.
"""
# Support people who used socket.settimeout() to escape
# handle_request before self.timeout was available.
timeout = self.socket.gettimeout()
if timeout is None:
timeout = self.timeout
elif self.timeout is not None:
timeout = min(timeout, self.timeout)
fd_sets = _eintr_retry(select.select, [self], [], [], timeout)
if not fd_sets[0]:
self.handle_timeout()
return
self._handle_request_noblock() def _handle_request_noblock(self):
"""Handle one request, without blocking. I assume that select.select has returned that the socket is
readable before this function was called, so there should be
no risk of blocking in get_request().
"""
try:
request, client_address = self.get_request()
except socket.error:
return
if self.verify_request(request, client_address):
try:
self.process_request(request, client_address)
except:
self.handle_error(request, client_address)
self.shutdown_request(request) def handle_timeout(self):
"""Called if no new request arrives within self.timeout. Overridden by ForkingMixIn.
"""
pass def verify_request(self, request, client_address):
"""Verify the request. May be overridden. Return True if we should proceed with this request. """
return True def process_request(self, request, client_address):
"""Call finish_request. Overridden by ForkingMixIn and ThreadingMixIn. """
self.finish_request(request, client_address)
self.shutdown_request(request) def server_close(self):
"""Called to clean-up the server. May be overridden. """
pass def finish_request(self, request, client_address):
"""Finish one request by instantiating RequestHandlerClass."""
self.RequestHandlerClass(request, client_address, self) def shutdown_request(self, request):
"""Called to shutdown and close an individual request."""
self.close_request(request) def close_request(self, request):
"""Called to clean up an individual request."""
pass def handle_error(self, request, client_address):
"""Handle an error gracefully. May be overridden. The default is to print a traceback and continue. """
print '-'*40
print 'Exception happened during processing of request from',
print client_address
import traceback
traceback.print_exc() # XXX But this goes to stderr!
print '-'*40BaseServer

class TCPServer(BaseServer): """Base class for various socket-based server classes. Defaults to synchronous IP stream (i.e., TCP). Methods for the caller: - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
- serve_forever(poll_interval=0.5)
- shutdown()
- handle_request() # if you don't use serve_forever()
- fileno() -> int # for select() Methods that may be overridden: - server_bind()
- server_activate()
- get_request() -> request, client_address
- handle_timeout()
- verify_request(request, client_address)
- process_request(request, client_address)
- shutdown_request(request)
- close_request(request)
- handle_error() Methods for derived classes: - finish_request(request, client_address) Class variables that may be overridden by derived classes or
instances: - timeout
- address_family
- socket_type
- request_queue_size (only for stream sockets)
- allow_reuse_address Instance variables: - server_address
- RequestHandlerClass
- socket """ address_family = socket.AF_INET socket_type = socket.SOCK_STREAM request_queue_size = 5 allow_reuse_address = False def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True):
"""Constructor. May be extended, do not override."""
BaseServer.__init__(self, server_address, RequestHandlerClass)
self.socket = socket.socket(self.address_family,
self.socket_type)
if bind_and_activate:
try:
self.server_bind()
self.server_activate()
except:
self.server_close()
raise def server_bind(self):
"""Called by constructor to bind the socket. May be overridden. """
if self.allow_reuse_address:
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind(self.server_address)
self.server_address = self.socket.getsockname() def server_activate(self):
"""Called by constructor to activate the server. May be overridden. """
self.socket.listen(self.request_queue_size) def server_close(self):
"""Called to clean-up the server. May be overridden. """
self.socket.close() def fileno(self):
"""Return socket file number. Interface required by select(). """
return self.socket.fileno() def get_request(self):
"""Get the request and client address from the socket. May be overridden. """
return self.socket.accept() def shutdown_request(self, request):
"""Called to shutdown and close an individual request."""
try:
#explicitly shutdown. socket.close() merely releases
#the socket and waits for GC to perform the actual close.
request.shutdown(socket.SHUT_WR)
except socket.error:
pass #some platforms may raise ENOTCONN here
self.close_request(request) def close_request(self, request):
"""Called to clean up an individual request."""
request.close()TCPServer

class ThreadingMixIn:
"""Mix-in class to handle each request in a new thread.""" # Decides how threads will act upon termination of the
# main process
daemon_threads = False def process_request_thread(self, request, client_address):
"""Same as in BaseServer but as a thread. In addition, exception handling is done here. """
try:
self.finish_request(request, client_address)
self.shutdown_request(request)
except:
self.handle_error(request, client_address)
self.shutdown_request(request) def process_request(self, request, client_address):
"""Start a new thread to process the request."""
t = threading.Thread(target = self.process_request_thread,
args = (request, client_address))
t.daemon = self.daemon_threads
t.start()

class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass

class BaseRequestHandler: """Base class for request handler classes. This class is instantiated for each request to be handled. The
constructor sets the instance variables request, client_address
and server, and then calls the handle() method. To implement a
specific service, all you need to do is to derive a class which
defines a handle() method. The handle() method can find the request as self.request, the
client address as self.client_address, and the server (in case it
needs access to per-server information) as self.server. Since a
separate instance is created for each request, the handle() method
can define arbitrary other instance variariables. """ def __init__(self, request, client_address, server):
self.request = request
self.client_address = client_address
self.server = server
self.setup()
try:
self.handle()
finally:
self.finish() def setup(self):
pass def handle(self):
pass def finish(self):
pass

#_*_coding:utf-8_*_
__author__ = 'Alex Li'import socket, logging
import select, errnologger = logging.getLogger("network-server")def InitLog():
logger.setLevel(logging.DEBUG) fh = logging.FileHandler("network-server.log")
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR) formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
ch.setFormatter(formatter)
fh.setFormatter(formatter) logger.addHandler(fh)
logger.addHandler(ch)if __name__ == "__main__":
InitLog() try:
# 创建 TCP socket 作为监听 socket
listen_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
except socket.error as msg:
logger.error("create socket failed") try:
# 设置 SO_REUSEADDR 选项
listen_fd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
except socket.error as msg:
logger.error("setsocketopt SO_REUSEADDR failed") try:
# 进行 bind -- 此处未指定 ip 地址，即 bind 了全部网卡 ip 上
listen_fd.bind(('', 2003))
except socket.error as msg:
logger.error("bind failed") try:
# 设置 listen 的 backlog 数
listen_fd.listen(10)
except socket.error as msg:
logger.error(msg) try:
# 创建 epoll 句柄
epoll_fd = select.epoll()
# 向 epoll 句柄中注册 监听 socket 的 可读 事件
epoll_fd.register(listen_fd.fileno(), select.EPOLLIN)
except select.error as msg:
logger.error(msg) connections = {}
addresses = {}
datalist = {}
while True:
# epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
epoll_list = epoll_fd.poll() for fd, events in epoll_list:
# 若为监听 fd 被激活
if fd == listen_fd.fileno():
# 进行 accept -- 获得连接上来 client 的 ip 和 port，以及 socket 句柄
conn, addr = listen_fd.accept()
logger.debug("accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno()))
# 将连接 socket 设置为 非阻塞
conn.setblocking(0)
# 向 epoll 句柄中注册 连接 socket 的 可读 事件
epoll_fd.register(conn.fileno(), select.EPOLLIN | select.EPOLLET)
# 将 conn 和 addr 信息分别保存起来
connections[conn.fileno()] = conn
addresses[conn.fileno()] = addr
elif select.EPOLLIN & events:
# 有 可读 事件激活
datas = ''
while True:
try:
# 从激活 fd 上 recv 10 字节数据
data = connections[fd].recv(10)
# 若当前没有接收到数据，并且之前的累计数据也没有
if not data and not datas:
# 从 epoll 句柄中移除该 连接 fd
epoll_fd.unregister(fd)
# server 侧主动关闭该 连接 fd
connections[fd].close()
logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
break
else:
# 将接收到的数据拼接保存在 datas 中
datas += data
except socket.error as msg:
# 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
# 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
if msg.errno == errno.EAGAIN:
logger.debug("%s receive %s" % (fd, datas))
# 将已接收数据保存起来
datalist[fd] = datas
# 更新 epoll 句柄中连接d 注册事件为 可写
epoll_fd.modify(fd, select.EPOLLET | select.EPOLLOUT)
break
else:
# 出错处理
epoll_fd.unregister(fd)
connections[fd].close()
logger.error(msg)
break
elif select.EPOLLHUP & events:
# 有 HUP 事件激活
epoll_fd.unregister(fd)
connections[fd].close()
logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
elif select.EPOLLOUT & events:
# 有 可写 事件激活
sendLen = 0
# 通过 while 循环确保将 buf 中的数据全部发送出去
while True:
# 将之前收到的数据发回 client -- 通过 sendLen 来控制发送位置
sendLen += connections[fd].send(datalist[fd][sendLen:])
# 在全部发送完毕后退出 while 循环
if sendLen == len(datalist[fd]):
break
# 更新 epoll 句柄中连接 fd 注册事件为 可读
epoll_fd.modify(fd, select.EPOLLIN | select.EPOLLET)
else:
# 其他 epoll 事件不进行处理
continueepoll socket echo server