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

用户11081884

发布于 2026-07-20 20:09:35

220

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

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

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

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

1. 异步HTTP请求

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

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

importaiohttp
importasyncio

asyncdeffetch(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实时通信

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

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

importasyncio
importwebsockets

asyncdefecho_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地址信息，帮你快速定位网络问题。

importsocket

defresolve_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协议发送电子邮件，实现自动化通知，再也不用手动发邮件了。

importsmtplib
fromemail.mime.textimportMIMEText

defsend_email(sender, receiver, password, subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = sender
    msg['To'] = receiver

    withsmtplib.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协议检测主机是否在线，快速判断网络是否畅通。

importsubprocess
importplatform

defping_host(host):
    param = '-n'ifplatform.system().lower() == 'windows'else'-c'
    command = ['ping', param, '3', host]
    returnsubprocess.call(command) == 0

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

6. TCP服务器与客户端

场景：自定义协议通信

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

# 服务端
importsocket

defstart_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文件下载

场景：从网络下载文件

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

importrequests

defdownload_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响应，轻松对接各种第三方服务。

importrequests

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

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

9. 网络接口信息获取

场景：系统网络配置查看

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

importnetifaces

defget_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证书有效性，保障通信安全。

importssl
importsocket
fromdatetimeimportdatetime

defcheck_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()).days
            print(f"SSL证书剩余有效期: {days_left}天")

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

11. 多进程端口扫描

场景：快速扫描多个端口

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

importsocket
fromconcurrent.futuresimportProcessPoolExecutor

defcheck_port(host, port):
    withsocket.socket(socket.AF_INET, socket.SOCK_STREAM) assock:
        sock.settimeout(1)
        result = sock.connect_ex((host, port))
        returnportifresult == 0elseNone

defscan_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和连接，轻松处理需要登录的操作。

importrequests

deflogin_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.allimportsniff

defpacket_callback(packet):
    ifpacket.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代理，突破网络限制。

importrequests

defthrough_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. 网络带宽测试

场景：网络性能测量

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

importrequests
importtime

defspeed_test(url, file_size_mb=10):
    start_time = time.time()
    response = requests.get(url, stream=True)

    total_size = 0
    forchunkinresponse.iter_content(chunk_size=1024):
        total_size += len(chunk)
        iftotal_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')

16. HTTP断点续传

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

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

importrequests
importos

defresume_download(url, filename):
    # 检查文件是否存在
    mode = 'ab'ifos.path.exists(filename) else'wb'
    downloaded = os.path.getsize(filename) ifos.path.exists(filename) else0

    headers = {'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是无连接协议，适合对实时性要求高的场景，牺牲一点可靠性换取速度。

importsocket

defudp_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缓存策略

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

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

importrequests
importtime

defcached_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连接，减少握手开销，提升高并发场景性能。

importaiohttp
importasyncio

asyncdeffetch_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)
        returnresults

asyncdeffetch_single(session, url):
    asyncwithsession.get(url) asresponse:
        returnawaitresponse.text()

# 使用示例
# urls = ['https://httpbin.org/json'] * 20
# results = asyncio.run(fetch_with_pool(urls))

20. 网络异常重试机制

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

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

importrequests
importtime
fromfunctoolsimportwraps

defretry_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:
                        raisee
                    print(f"重试第 {attempt + 1} 次...")
                    time.sleep(delay* (attempt+1))
        returnwrapper
    returndecorator

@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为此提供了强大的支持。在实际应用中，请始终遵守网络礼仪和相关法律法规。

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

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

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

原始发表：2026-03-23，如有侵权请联系 cloudcommunity@tencent.com 删除

效率