Python 20个网络编程方法,让你的网络操作效率暴涨300%

你是不是也遇到过这些情况？

每次处理网络请求时都要等半天，并发一多就卡得不行？想实现个实时通信功能，结果被各种协议搞得头大？

别担心，今天给你整理了20个超实用的Python网络编程方法，从基础请求到高级应用，帮你轻松搞定网络编程！

往期Python阅读>>

Python 开发中常见的高效写法

Python 快速开发Web文件服务器

Python 15个自动化获取金融数据接口库

Python 自动化做数据可视化10个示例（含代码），强烈推荐

Python 20个实用高效装饰器

Python 30个操作系统命令，事半功倍

Python 40个实用代码案例：提升效率

Python Webbrowser自动化控制浏览器窗口

Python pathlib的使用方法

Python 自动化监控大文件

Python 90个经典使用技巧

Python 文件监控利器Watchdog的使用方法

Docker 40个自动化管理脚本

Python 回调函数的常见应用场景

Python 40个常见场景的代码示例

Python 20 个常用标准模块

Python requests库用法详解

Python 25个函数的开发技巧

Python 20个提高学习效率的工具

Python 自动化文件管理的10个模板

Python 20个常用的开发库

Python 20个代码优化方法

Python 30个内置函数全面解析

Python 50个命令行技巧

Python 解析与转换JSON格式

1. 异步HTTP请求

场景：高效处理大量并发网络请求 

使用异步IO提升请求效率，适合高并发场景，再也不用等待每个请求依次完成了。

importaiohttpimportasyncioasyncdeffetch(url):asyncwithaiohttp.ClientSession() assession:asyncwithsession.get(url) asresponse:returnawaitresponse.text()asyncdefmain():url = "https://httpbin.org/json"html = awaitfetch(url)print(html[:200])asyncio.run(main())

2. WebSocket实时通信

场景：实时聊天、股票行情推送 

建立双向通信通道，实现服务器主动推送，再也不用轮询了。

importasyncioimportwebsocketsasyncdefecho_client():asyncwithwebsockets.connect('ws://localhost:8765') aswebsocket:awaitwebsocket.send("Hello WebSocket!")response = awaitwebsocket.recv()print(f"Received: {response}")asyncio.run(echo_client())

3. DNS解析查询

场景：域名解析、网络诊断 

获取域名的IP地址信息，帮你快速定位网络问题。

importsocketdefresolve_dns(domain):try:ip_list = socket.getaddrinfo(domain, None)forresultinip_list:print(f"IP: {result[0]}")exceptsocket.gaierrorase:print(f"DNS resolution failed: {e}")resolve_dns("github.com")

4. SMTP邮件发送

场景：自动发送通知邮件 

通过SMTP协议发送电子邮件，实现自动化通知，再也不用手动发邮件了。

importsmtplibfromemail.mime.textimportMIMETextdefsend_email(sender, receiver, password, subject, body):msg = MIMEText(body)msg['Subject'] = subjectmsg['From'] = sendermsg['To'] = receiverwithsmtplib.SMTP('smtp.gmail.com', 587) asserver:server.starttls()server.login(sender, password)server.send_message(msg)# 使用示例（需配置真实邮箱）# send_email('sender@email.com', 'receiver@email.com',#           'password', 'Test Subject', 'Hello from Python!')

5. ICMP Ping检测

场景：网络连通性测试 

使用ICMP协议检测主机是否在线，快速判断网络是否畅通。

importsubprocessimportplatformdefping_host(host):param = '-n'ifplatform.system().lower() == 'windows'else'-c'command = ['ping', param, '3', host]returnsubprocess.call(command) == 0ifping_host('8.8.8.8'):print("Host is reachable")else:print("Host is unreachable")

6. TCP服务器与客户端

场景：自定义协议通信 

实现基础的TCP服务端和客户端，轻松搭建自己的通信服务。

# 服务端importsocketdefstart_tcp_server(host='localhost', port=8888):withsocket.socket(socket.AF_INET, socket.SOCK_STREAM) ass:s.bind((host, port))s.listen()print(f"Server listening on {host}:{port}")conn, addr = s.accept()withconn:data = conn.recv(1024)conn.sendall(b'Server received: '+data)# 客户端deftcp_client(message, host='localhost', port=8888):withsocket.socket(socket.AF_INET, socket.SOCK_STREAM) ass:s.connect((host, port))s.sendall(message.encode())data = s.recv(1024)print(f"Received: {data.decode()}")

7. HTTP文件下载

场景：从网络下载文件 

分段下载大文件，支持进度显示，再也不怕下载中断了。

importrequestsdefdownload_file(url, local_filename):withrequests.get(url, stream=True) asr:r.raise_for_status()withopen(local_filename, 'wb') asf:forchunkinr.iter_content(chunk_size=8192):f.write(chunk)returnlocal_filename# download_file('https://example.com/largefile.zip', 'downloaded.zip')

8. REST API调用

场景：与Web API交互 

调用RESTful接口并处理JSON响应，轻松对接各种第三方服务。

importrequestsdefget_weather(city):# 示例API（需替换为真实API端点）url = f"https://api.weather.com/v1/{city}/current"response = requests.get(url)ifresponse.status_code == 200:returnresponse.json()else:returnNoneweather_data = get_weather("beijing")print(weather_data)

9. 网络接口信息获取

场景：系统网络配置查看 

获取本机网络接口详细信息，快速了解网络配置。

importnetifacesdefget_network_info():interfaces = netifaces.interfaces()forinterfaceininterfaces:addrs = netifaces.ifaddresses(interface)ifnetifaces.AF_INETinaddrs:forlinkinaddrs[netifaces.AF_INET]:print(f"Interface: {interface}")print(f"IP Address: {link['addr']}")print(f"Netmask: {link['netmask']}")get_network_info()

10. SSL证书验证

场景：安全连接验证 

检查网站SSL证书有效性，保障通信安全。

importsslimportsocketfromdatetimeimportdatetimedefcheck_ssl_cert(hostname):context = ssl.create_default_context()withsocket.create_connection((hostname, 443)) assock:withcontext.wrap_socket(sock, server_hostname=hostname) asssock:cert = ssock.getpeercert()expire_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')days_left = (expire_date-datetime.now()).daysprint(f"SSL证书剩余有效期: {days_left}天")check_ssl_cert("https://www.google.com")

11. 多进程端口扫描

场景：快速扫描多个端口 

使用多进程加速端口扫描，大幅提升检测效率。

importsocketfromconcurrent.futuresimportProcessPoolExecutordefcheck_port(host, port):withsocket.socket(socket.AF_INET, socket.SOCK_STREAM) assock:sock.settimeout(1)result = sock.connect_ex((host, port))returnportifresult == 0elseNonedefscan_ports(host, start_port=1, end_port=100):withProcessPoolExecutor() asexecutor:results = executor.map(check_port, [host]* (end_port-start_port),range(start_port, end_port))open_ports = [portforportinresultsifport]print(f"开放端口: {open_ports}")scan_ports('localhost', 80, 90)

12. HTTP请求会话保持

场景：需要登录状态的网络操作 

使用Session对象维持Cookie和连接，轻松处理需要登录的操作。

importrequestsdeflogin_session(username, password):session = requests.Session()# 模拟登录login_data = {'username': username, 'password': password}session.post('https://example.com/login', data=login_data)# 使用会话访问需要认证的页面profile = session.get('https://example.com/profile')returnprofile.content# html_content = login_session('user', 'pass')

13. 网络数据包嗅探

场景：网络流量分析 

使用scapy库捕获和分析网络包，深入了解网络通信。

fromscapy.allimportsniffdefpacket_callback(packet):ifpacket.haslayer('IP'):ip_src = packet['IP'].srcip_dst = packet['IP'].dstprint(f"Packet: {ip_src} -> {ip_dst}")# 捕获10个数据包（需要管理员权限）# sniff(prn=packet_callback, count=10)

14. 代理服务器使用

场景：通过代理访问网络 

配置HTTP/HTTPS代理，突破网络限制。

importrequestsdefthrough_proxy(url, proxy_url):proxies = {'http': proxy_url,'https': proxy_url    }response = requests.get(url, proxies=proxies, timeout=10)returnresponse.status_code# status = through_proxy('https://httpbin.org/ip', 'http://proxy-server:8080')

15. 网络带宽测试

场景：网络性能测量 

通过下载测试估算带宽，了解你的网络速度。

importrequestsimporttimedefspeed_test(url, file_size_mb=10):start_time = time.time()response = requests.get(url, stream=True)total_size = 0forchunkinresponse.iter_content(chunk_size=1024):total_size += len(chunk)iftotal_size>= file_size_mb*1024*1024:breakend_time = time.time()speed = (total_size/ (end_time-start_time)) /1024/1024print(f"下载速度: {speed:.2f} MB/s")# speed_test('https://speedtest.tele2.net/100MB.zip')

16. HTTP断点续传

场景：大文件下载中断恢复 

支持断点续传，下载中断后可以继续下载，节省时间和流量。

importrequestsimportosdefresume_download(url, filename):# 检查文件是否存在mode = 'ab'ifos.path.exists(filename) else'wb'downloaded = os.path.getsize(filename) ifos.path.exists(filename) else0headers = {'Range': f'bytes={downloaded}-'}withrequests.get(url, headers=headers, stream=True) asr:r.raise_for_status()withopen(filename, mode) asf:forchunkinr.iter_content(chunk_size=8192):f.write(chunk)# resume_download('https://example.com/largefile.zip', 'resume_file.zip')

17. UDP服务器与客户端

场景：实时数据传输、视频流 

UDP是无连接协议，适合对实时性要求高的场景，牺牲一点可靠性换取速度。

importsocketdefudp_server(host='localhost', port=9999):withsocket.socket(socket.AF_INET, socket.SOCK_DGRAM) ass:s.bind((host, port))print(f"UDP Server listening on {host}:{port}")whileTrue:data, addr = s.recvfrom(1024)print(f"Received from {addr}: {data.decode()}")s.sendto(b"ACK", addr)defudp_client(message, host='localhost', port=9999):withsocket.socket(socket.AF_INET, socket.SOCK_DGRAM) ass:s.sendto(message.encode(), (host, port))data, _ = s.recvfrom(1024)print(f"Server response: {data.decode()}")

18. HTTP缓存策略

场景：减少重复请求，提升性能 

合理使用缓存头，减少不必要的网络请求，提升响应速度。

importrequestsimporttimedefcached_request(url, cache_time=3600):cache_file = f"cache_{hash(url)}.txt"# 检查缓存是否存在且未过期ifos.path.exists(cache_file):file_time = os.path.getmtime(cache_file)iftime.time() -file_time<cache_time:withopen(cache_file, 'r') asf:returnf.read()# 发起新请求response = requests.get(url)withopen(cache_file, 'w') asf:f.write(response.text)returnresponse.text

19. 异步TCP连接池

场景：高并发长连接管理 

使用连接池复用TCP连接，减少握手开销，提升高并发场景性能。

importaiohttpimportasyncioasyncdeffetch_with_pool(urls):connector = aiohttp.TCPConnector(limit=10, force_close=False)asyncwithaiohttp.ClientSession(connector=connector) assession:tasks = [fetch_single(session, url) forurlinurls]results = awaitasyncio.gather(*tasks)returnresultsasyncdeffetch_single(session, url):asyncwithsession.get(url) asresponse:returnawaitresponse.text()# 使用示例# urls = ['https://httpbin.org/json'] * 20# results = asyncio.run(fetch_with_pool(urls))

20. 网络异常重试机制

场景：提升网络请求稳定性 

自动重试失败的请求，避免因临时网络波动导致任务失败。

importrequestsimporttimefromfunctoolsimportwrapsdefretry_request(max_retries=3, delay=1):defdecorator(func):@wraps(func)defwrapper(*args, **kwargs):forattemptinrange(max_retries):try:returnfunc(*args, **kwargs)exceptExceptionase:ifattempt == max_retries-1:raiseeprint(f"重试第 {attempt + 1} 次...")time.sleep(delay* (attempt+1))returnwrapperreturndecorator@retry_request(max_retries=3, delay=2)deffetch_with_retry(url):response = requests.get(url, timeout=5)returnresponse.text# 使用示例# content = fetch_with_retry('https://httpbin.org/json')

通过这20个网络编程方法，你已经掌握了从基础通信到高级应用的完整技能栈。

网络编程的核心在于理解协议原理和合理选择工具，Python为此提供了强大的支持。在实际应用中，请始终遵守网络礼仪和相关法律法规。

注意：部分代码需要安装额外依赖库，且涉及网络操作时请确保拥有相应权限。

想高效学习Python？下面三本精选好书满足你的不同需求！

《流畅的Python（第2版）》——Python进阶必读！深入讲解高级特性与最佳实践，适合想精进的开发者。

《Python从新手到高手》：初学者首选，系统学习全栈技能。

《Python数据分析：从零基础入门到案例实战》——数据科学利器！手把手教你用Python处理数据，实战案例学完就能用。

三本书均支持先用后付、运费险和7天无理由退货，放心购买！点击“购买”按钮，立即开启你的Python学习之旅吧！

https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c