Maison >développement back-end >Tutoriel Python >codes communs de programmation réseau python
Code côté serveur :
# -*- coding: cp936 -*- import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)#初始化socket sock.bind(("127.0.0.1", 8001))#绑定本机地址,8001端口 sock.listen(5)#等待客户连接 while True: print "waiting client connection..." connection,address = sock.accept()#接收客户连接请求 print "a client have connected..." while True: try: connection.settimeout(5) #设置超时时间 buf = connection.recv(1024) #接收数据 if buf == "1": connection.send("you have send me 1!welcome to server!") elif buf=="2": connection.send("you have send me 2!I have recv!") elif buf=="3": connection.send("close the connection!") break else: connection.send("unknow command!") except socket.timeout: print "time out" connection.close() print "a client exit..."
Code côté client :
import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(("127.0.0.1", 8001)) import time time.sleep(2) while True: data=raw_input("input command:"); sock.send(data) print sock.recv(1024) if data=="3": break sock.close()
1. Ouvrez d'abord deux IDLE et ouvrez respectivement les codes côté serveur et côté client.
2.F5 exécute le code côté serveur et la connexion client en attente apparaîtra...
3.F5 exécute le code côté client et la commande d'entrée : apparaît
4. le serveur et le client sont connectés. Désormais, une communication normale est possible, comme le montre la figure :
5. Une erreur se produira lors de la nouvelle exécution du code côté serveur. cette fois, vous pouvez utiliser le gestionnaire de tâches pour modifier le pythonw. Une fois le processus .exe terminé, rouvrez-le et compilez-le !
Client :
import socket s=socket.socket() host = socket.gethostname() port = 1234 s.connect((host, port)) print s.recv(1024)
Serveur :
import socket s = socket.socket() host = socket.gethostname() port = 1234 s.bind((host, port)) s.listen(5) while True: c, addr = s.accept() print 'Got connection from', addr c.send('Thank you for connecting') c.close()
programmation http
from urllib import urlopen webpage = urlopen('http://www.python.org')
plus expressions régulières
import re text = webpage.read() m = re.search('<a href="([^"]+)" .*?>about</a>', text, re.IGNORECASE) m.group(1)
urllib
urllib2
Un petit exemple de serveur basé sur SocketServer :
from SocketServer import TCPServer, StreamRequestHandler >>> class Handler(StreamRequestHandler): def handle(self): addr = self.request.getpeername() print 'Got connection from', addr self.wfile.write('Thank you for connecting') >>> server = TCPServer(('',1234), Handler) >>> server.serve_forever()
Utilisation de SocketServer pour le forking et le threading Gestion
Fork Serveur :
from SocketServer import TCPServer,ForkingMinIn, StreamRequestHandler class Server(ForkingMinIn, TCPServer):pass class Handler(StreamRequestHandler): def handle(self): addr = self.request.getpeername() print 'Got connection from',addr self.wfile.write('Thank you for connection') server = Server(('',1234),Handler) server.serve_forever()
Serveur de thread :
from SocketServer import TCPServer, ThreadingMixIn, StreamRequestHandler class Server(ThreadingMixIn, TCPServer):pass class Handler(StreamRequestHandler): def handle(self): addr = self.request.getpeername() print 'Got connection from',addr self.wfile.write('Thank you for connecting') server = Server(('',1234),Handler) server.serve_forever()
I/Asynchrone O avec service select et pool
select :
import socket, select s = socket.socket() host = socket.gethostname() prot = 1234 s.bind((host,port)) s.listen(5) inputs = [s] while True: rs,ws,es = select.select(inputs,[],[]) for r in rs: if r is s: c, addr = s.accept() print 'Got connection from',addr inputs.append(c) else: try: data = r.recv(1024) disconnected = not data except socket.error: disconnected = True if disconnected: print r.getpeername(), 'disconnected' inputs.remove(r) else: print data
serveur de pool :
import socket, select s = socket.socket() host = socket.gethostname() port = 1234 s.bind((host, port)) fdmap = {s.fileno():s} s.listen(5) p = select.poll() p.register(s) while True: events = p.poll() for fd, event in events: if fd in fdmap: c, addr = s.accept() print 'Got connection from', addr p.register(c) fdmap[c.fileno()]=c elif event & select.POLLIN: data = fdmap[fd].recv(1024) if not data: print fdmap[fd].getpeername(),'disconnected' p.unregister(fd) del fdmap[fd] else: print data
Pour plus d'articles liés aux codes de programmation réseau Python courants, veuillez faire attention au site Web PHP chinois !