Python 15个网络编程方法

用户11081884

发布于 2026-07-20 19:13:34

890

文章被收录于专栏：用户11081884的专栏用户11081884的专栏

Python凭借其简洁的语法和丰富的库成为网络编程的首选语言之一。本文将通过15个实用方法，从基础请求到高级应用，帮助掌握Python网络编程的核心技能。

1. 异步HTTP请求（aiohttp）

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

说明：使用异步IO提升请求效率，适合高并发场景

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    url = " https://httpbin.org/json "
    html = await fetch(url)
    print(html[:200])

asyncio.run(main())

2. WebSocket实时通信

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

说明：建立双向通信通道，实现服务器主动推送

import asyncio
import websockets

async def echo_client():
    async with websockets.connect('ws://localhost:8765') as websocket:
        await websocket.send("Hello WebSocket!")
        response = await websocket.recv()
        print(f"Received: {response}")

asyncio.run(echo_client())

3. DNS解析查询

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

说明：获取域名的IP地址信息

import socket

def resolve_dns(domain):
    try:
        ip_list = socket.getaddrinfo(domain, None)
        for result in ip_list:
            print(f"IP: {result[0]}")
    except socket.gaierror as e:
        print(f"DNS resolution failed: {e}")

resolve_dns("github.com")

4. SMTP邮件发送

场景：自动发送通知邮件

说明：通过SMTP协议发送电子邮件

import smtplib
from email.mime.text import MIMEText

def send_email(sender, receiver, password, subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = sender
    msg['To'] = receiver
    
    with smtplib.SMTP('smtp.gmail.com', 587) as server:
        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协议检测主机是否在线

import subprocess
import platform

def ping_host(host):
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    command = ['ping', param, '3', host]
    return subprocess.call(command) == 0

if ping_host('8.8.8.8'):
    print("Host is reachable")
else:
    print("Host is unreachable")

6. TCP服务器与客户端

场景：自定义协议通信

说明：实现基础的TCP服务端和客户端

# 服务端
import socket

def start_tcp_server(host='localhost', port=8888):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((host, port))
        s.listen()
        print(f"Server listening on {host}:{port}")
        
        conn, addr = s.accept()
        with conn:
            data = conn.recv(1024)
            conn.sendall(b'Server received: ' + data)

# 客户端
def tcp_client(message, host='localhost', port=8888):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((host, port))
        s.sendall(message.encode())
        data = s.recv(1024)
        print(f"Received: {data.decode()}")

7. HTTP文件下载

场景：从网络下载文件

说明：分段下载大文件，支持进度显示

import requests

def download_file(url, local_filename):
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(local_filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
    return local_filename

# download_file(' https://example.com/largefile.zip', 'downloaded.zip')

8. REST API调用

场景：与Web API交互

说明：调用RESTful接口并处理JSON响应

import requests

def get_weather(city):
    # 示例API（需替换为真实API端点）
    url = f" https://api.weather.com/v1/ {city}/current"
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

weather_data = get_weather("beijing")
print(weather_data)

9. 网络接口信息获取

场景：系统网络配置查看

说明：获取本机网络接口详细信息

import netifaces

def get_network_info():
    interfaces = netifaces.interfaces()
    for interface in interfaces:
        addrs = netifaces.ifaddresses(interface)
        if netifaces.AF_INET in addrs:
            for link in addrs[netifaces.AF_INET]:
                print(f"Interface: {interface}")
                print(f"IP Address: {link['addr']}")
                print(f"Netmask: {link['netmask']}")

get_network_info()

10. SSL证书验证

场景：安全连接验证

说明：检查网站SSL证书有效性

import ssl
import socket
from datetime import datetime

def check_ssl_cert(hostname):
    context = ssl.create_default_context()
    with socket.create_connection((hostname, 443)) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            cert = ssock.getpeercert()
            expire_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
            days_left = (expire_date - datetime.now()).days
            print(f"SSL证书剩余有效期: {days_left}天")

check_ssl_cert(" https://www.google.com ")

11. 多进程端口扫描

场景：快速扫描多个端口

说明：使用多进程加速端口扫描

import socket
from concurrent.futures import ProcessPoolExecutor

def check_port(host, port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.settimeout(1)
        result = sock.connect_ex((host, port))
        return port if result == 0 else None

def scan_ports(host, start_port=1, end_port=100):
    with ProcessPoolExecutor() as executor:
        results = executor.map(check_port, [host]* (end_port-start_port), 
                             range(start_port, end_port))
    
    open_ports = [port for port in results if port]
    print(f"开放端口: {open_ports}")

scan_ports('localhost', 80, 90)

12. HTTP请求会话保持

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

说明：使用Session对象维持Cookie和连接

import requests

def login_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')
    return profile.content

# html_content = login_session('user', 'pass')

13. 网络数据包嗅探

场景：网络流量分析

说明：使用scapy库捕获和分析网络包

from scapy.all import sniff

def packet_callback(packet):
    if packet.haslayer('IP'):
        ip_src = packet['IP'].src
        ip_dst = packet['IP'].dst
        print(f"Packet: {ip_src} -> {ip_dst}")

# 捕获10个数据包（需要管理员权限）
# sniff(prn=packet_callback, count=10)

14. 代理服务器使用

场景：通过代理访问网络

说明：配置HTTP/HTTPS代理

import requests

def through_proxy(url, proxy_url):
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    response = requests.get(url, proxies=proxies, timeout=10)
    return response.status_code

# status = through_proxy(' https://httpbin.org/ip', ' http://proxy-server:8080')

15. 网络带宽测试

场景：网络性能测量

说明：通过下载测试估算带宽

import requests
import time

def speed_test(url, file_size_mb=10):
    start_time = time.time()
    response = requests.get(url, stream=True)
    
    total_size = 0
    for chunk in response.iter_content(chunk_size=1024):
        total_size += len(chunk)
        if total_size >= file_size_mb * 1024 * 1024:
            break
    
    end_time = time.time()
    speed = (total_size / (end_time - start_time)) / 1024 / 1024
    print(f"下载速度: {speed:.2f} MB/s")

# speed_test(' https://speedtest.tele2.net/100MB.zip')

网络编程方法从基础通信到高级应用的完整技能栈。网络编程的核心在于理解协议原理和合理选择工具，Python为此提供了强大的支持。

“无他，惟手熟尔”！有需要的用起来！

如果你觉得这篇文章有用，欢迎点赞、转发、收藏、留言、推荐❤！

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-12-17，如有侵权请联系 cloudcommunity@tencent.com 删除

python