当前位置： 首页 > news >正文

北京做网站源代码的传媒公司做网站编辑_如何?

news 2025/10/12 19:22:46
北京做网站源代码的,传媒公司做网站编辑_如何?,网站站群 硬盘扩容 申请报告,wordpress 插件分类一、前言 引入异步编程趋势：Python的异步编程正变得越来越流行。在过去，同步的HTTP请求已经不足以满足对性能的要求。异步HTTP客户端库的流行：目前，有许多第三方库已经实现了异步HTTP客户端，如aiohttp和httpx等。然而…

一、前言

  • 引入异步编程趋势：Python的异步编程正变得越来越流行。在过去，同步的HTTP请求已经不足以满足对性能的要求。
  • 异步HTTP客户端库的流行：目前，有许多第三方库已经实现了异步HTTP客户端，如aiohttp和httpx等。然而，异步语法使得代码变得更加冗长，导致缩进增多，降低了代码的可读性和简洁性。
  • 封装异步HTTP客户端：为了简化异步HTTP请求的代码，我们需要封装一个常用的HTTP客户端，以实现业务中常见的功能，并提供更简洁的接口。在这篇博客中，我将使用httpx库来进行封装异步客户端，requests则是封装同步客户端，以实现常见的HTTP方法，并支持设置超时时间、请求参数等功能。

原文：Python 同、异步HTTP客户端封装：性能与简洁性的较量

二、同异步http客户端测试

同异步简易Demo

再封装之前先看看同异步发个http请求的代码差异，这里以 requests、aiohttp、httpx进行展示

依赖安装

pip install requests aiohttp httpx  

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 模块描述 }
# @Date: 2023/09/28 10:09
import asyncio
import httpx
import aiohttp
import requestsdef requests_demo(url):print("requests_demo")resp = requests.get(url)print(resp.text)async def aiohttp_demo(url):print("aiohttp_demo")async with aiohttp.client.ClientSession() as session:async with session.get(url) as resp:html_text = await resp.text()print(html_text)async def httpx_demo(url):print("httpx_demo")async with httpx.AsyncClient() as client:resp = await client.get(url)print(resp.text)async def main():url = "https://juejin.cn/"requests_demo(url)await aiohttp_demo(url)await httpx_demo(url)if __name__ == '__main__':asyncio.run(main())

可以看到同步的requests库实现的非常简洁一行代码就可以发送http请求。但异步语法的 httpx与aiohttp就感觉代码很臃肿，要嵌套好多层，尤其aiohttp，可读性变差了好多，但异步的请求可以大大的提升并发性能，利用网络IO的耗时处理更多的请求任务，这在爬虫中可以大大提升性能，再异步的web框架中也非常适用。

并发http请求测试

再看看同异步如何并发请求数据

async def concurrent_http_test():# requests testurls = ["https://juejin.cn/"] * 10start_time = time.time()for url in urls:requests_demo(url)use_time = time.time() - start_timeprint(f"requests {len(urls)} http req use {use_time} s")# httpx teststart_time = time.time()await asyncio.gather(*[httpx_demo(url) for url in urls])use_time = time.time() - start_timeprint(f"httpx {len(urls)} http req use {use_time} s")# aiohttp teststart_time = time.time()await asyncio.gather(*[aiohttp_demo(url) for url in urls])use_time = time.time() - start_timeprint(f"aiohttp {len(urls)} http req use {use_time} s")

结果：

requests 10 http req use 2.9108400344848633 shttpx 10 http req use 0.8657052516937256 saiohttp 10 http req use 1.9703822135925293 s

requests 请求demo是同步一个一个请求，所以会慢好多，而 httpx、aiohttp 是通过 asyncio.gather 并发请求的，会一次性发送10个请求，这样网络IO的耗时就复用了，但发现 aiohttp 的效果不尽人意，与httpx的0.86s相差太大，都是异步库，不应该的，于是看看之前写的demo代码发现其实aiohttp并没有复用 ClientSession 每次都是创建一个新的实例来去发送请求，这样频繁的创建与销毁连接会大大影响性能，httpx的 async with httpx.AsyncClient() as client: 好像是一样的问题，但httpx效果更好些。

尝试把 aiohttp 的 ClientSession 与 httpx.AsyncClient() 放到全局中去，再试试。

def requests_demo(url, session):# print("requests_demo")resp = session.get(url)return respasync def aiohttp_demo(url, aio_session):# print("aiohttp_demo")async with aio_session.get(url) as resp:# html_text = await resp.text()return respasync def httpx_demo(url, client):# print("httpx_demo")resp = await client.get(url)return respasync def concurrent_http_test():# requests testurls = ["https://juejin.cn/"] * 10start_time = time.time()with ThreadPoolExecutor() as pool:session = requests.session()for url in urls:pool.submit(requests_demo, url, session)use_time = time.time() - start_timeprint(f"requests {len(urls)} http req use {use_time} s")# aiohttp teststart_time = time.time()async with aiohttp.client.ClientSession() as aio_session:await asyncio.gather(*[aiohttp_demo(url, aio_session) for url in urls])use_time = time.time() - start_timeprint(f"aiohttp {len(urls)} http req use {use_time} s")# httpx teststart_time = time.time()async with httpx.AsyncClient() as client:await asyncio.gather(*[httpx_demo(url, client) for url in urls])use_time = time.time() - start_timeprint(f"httpx {len(urls)} http req use {use_time} s")

改进效果

requests 10 http req use 1.2176601886749268 saiohttp 10 http req use 0.4052879810333252 shttpx 10 http req use 0.5238490104675293 s

异步的效果很明显快了很多，requests 请求我也用 session 与线程池来并发请求看看效果，但网络有波动每次测的数据都不一样，所以这里的测试值仅作为参考。

三、异步http客户端封装

简易封装

aiohttp 与 httpx 性能都差不多，由于之前用 requests 习惯了，再接触这些异步封装的语法都觉得好怪，而 httpx的api 与 requests 类似，所以我就选择用 htppx 简单封装下。

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { http客户端 }
# @Date: 2023/08/10 09:33
import httpx
from datetime import timedeltaclass HttpMethod(BaseEnum):GET = "GET"POST = "POST"PATCH = "PATCH"PUT = "PUT"DELETE = "DELETE"HEAD = "HEAD"OPTIONS = "OPTIONS"class RespFmt(BaseEnum):"""http响应格式"""JSON = "json"BYTES = "bytes"TEXT = "text"class AsyncHttpClient:"""异步HTTP客户端通过httpx封装，实现了常见的HTTP方法,支持设置超时时间、请求参数等，简化了异步调用的层级缩进。Attributes:default_timeout: 默认请求超时时间,单位秒default_headers: 默认请求头字典default_resp_fmt: 默认响应格式jsonclient: httpx 异步客户端response: 每次实例请求的响应"""def __init__(self, timeout=timedelta(seconds=10), headers: dict = None, resp_fmt: RespFmt = RespFmt.JSON):"""构造异步HTTP客户端"""self.default_timeout = timeoutself.default_headers = headers or {}self.default_resp_fmt = resp_fmtself.client = httpx.AsyncClient()self.response: httpx.Response = Noneasync def _request(self,method: HttpMethod, url: str,params: dict = None, data: dict = None,timeout: timedelta = None, **kwargs):"""内部请求实现方法创建客户端会话,构造并发送HTTP请求,返回响应对象Args:method: HttpMethod 请求方法, 'GET', 'POST' 等url: 请求URLparams: 请求查询字符串参数字典data: 请求体数据字典timeout: 超时时间,单位秒kwargs: 其他关键字参数Returns:httpx.Response: HTTP响应对象"""timeout = timeout or self.default_timeoutheaders = self.default_headers or {}self.response = await self.client.request(method=method.value,url=url,params=params,data=data,headers=headers,timeout=timeout.total_seconds(),**kwargs)return self.responsedef _parse_response(self, resp_fmt: RespFmt = None):"""解析响应Args:resp_fmt: 响应格式Returns:resp Union[dict, bytes, str]"""resp_fmt = resp_fmt or self.default_resp_fmtresp_content_mapping = {RespFmt.JSON: self.json,RespFmt.BYTES: self.bytes,RespFmt.TEXT: self.text,}resp_func = resp_content_mapping.get(resp_fmt)return resp_func()def json(self):return self.response.json()def bytes(self):return self.response.contentdef text(self):return self.response.textasync def get(self, url: str, params: dict = None, timeout: timedelta = None, resp_fmt: RespFmt = None, **kwargs):"""GET请求Args:url: 请求URLparams: 请求查询字符串参数字典timeout: 请求超时时间,单位秒resp_fmt: 响应格式，默认None 使用实例对象的 default_resp_fmtReturns:resp => dict or bytes"""await self._request(HttpMethod.GET, url, params=params, timeout=timeout, **kwargs)return self._parse_response(resp_fmt)async def post(self, url: str, data: dict = None, timeout: timedelta = None, resp_fmt: RespFmt = None, **kwargs):"""POST请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒resp_fmt: 响应格式，默认None 使用实例对象的 default_resp_fmtReturns:resp => dict or bytes"""await self._request(HttpMethod.POST, url, data=data, timeout=timeout, **kwargs)return self._parse_response(resp_fmt)async def put(self, url: str, data: dict = None, timeout: timedelta = None, resp_fmt: RespFmt = None, **kwargs):"""PUT请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒resp_fmt: 响应格式，默认None 使用实例对象的 default_resp_fmtReturns:resp => dict"""await self._request(HttpMethod.PUT, url, data=data, timeout=timeout, **kwargs)return self._parse_response(resp_fmt)async def delete(self, url: str, data: dict = None, timeout: timedelta = None, resp_fmt: RespFmt = None, **kwargs):"""DELETE请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒resp_fmt: 响应格式，默认None 使用实例对象的 default_resp_fmtReturns:resp => dict"""await self._request(HttpMethod.DELETE, url, data=data, timeout=timeout, **kwargs)return self._parse_response(resp_fmt)

封装细节

这里封装就是简单的内部维护一个 httpx 的异步客户端，然后初始化一些默认的参数

  • default_timeout: 默认请求超时时间，单位秒，默认10s
  • default_headers: 默认请求头字典
  • default_resp_fmt: 默认响应格式json
  • client: httpx 异步客户端
  • response: 每次实例请求的响应

class AsyncHttpClient:"""异步HTTP客户端"""def __init__(self, timeout=timedelta(seconds=10), headers: dict = None, resp_fmt: RespFmt = RespFmt.JSON):"""构造异步HTTP客户端"""self.default_timeout = timeoutself.default_headers = headers or {}self.default_resp_fmt = resp_fmtself.client = httpx.AsyncClient()self.response: httpx.Response = None

然后实现几个常用的请求，get、post、put、delete方法

async def post(self, url: str, data: dict = None, timeout: timedelta = None, resp_fmt: RespFmt = None, **kwargs):"""POST请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒resp_fmt: 响应格式，默认None 使用实例对象的 default_resp_fmtReturns:resp => dict or bytes"""await self._request(HttpMethod.POST, url, data=data, timeout=timeout, **kwargs)return self._parse_response(resp_fmt)

每个请求方法冗余了一些常用的参数字段，例如

  • params 查询字符串入参

  • data body入参

  • timeout: 请求超时时间,单位秒

  • resp_fmt: 响应格式，默认None 使用实例对象的 default_resp_fmt

    • 默认json，一般我们http的数据交互都是使用 json了

  • **kwargs 预留其他关键字参数的入参

    • 这样有助于有些参数没想到要设计但经常用，可以通过kwargs来弥补

其实 get、post、put、delete方法没做什么事，就是标记了下使用什么请求方法、参数，最终都是让 _request方法处理。

async def _request(self,method: HttpMethod, url: str,params: dict = None, data: dict = None,timeout: timedelta = None, **kwargs
):"""内部请求实现方法创建客户端会话,构造并发送HTTP请求,返回响应对象Args:method: HttpMethod 请求方法, 'GET', 'POST' 等url: 请求URLparams: 请求查询字符串参数字典data: 请求体数据字典timeout: 超时时间,单位秒kwargs: 其他关键字参数Returns:httpx.Response: HTTP响应对象"""timeout = timeout or self.default_timeoutheaders = self.default_headers or {}self.response = await self.client.request(method=method.value,url=url,params=params,data=data,headers=headers,timeout=timeout.total_seconds(),**kwargs)return self.response

处理完再根据指定的响应格式进行解析

def _parse_response(self, resp_fmt: RespFmt = None):"""解析响应Args:resp_fmt: 响应格式Returns:resp Union[dict, bytes, str]"""resp_fmt = resp_fmt or self.default_resp_fmtresp_content_mapping = {RespFmt.JSON: self.json,RespFmt.BYTES: self.bytes,RespFmt.TEXT: self.text,}resp_func = resp_content_mapping.get(resp_fmt)return resp_func()def json(self):return self.response.json()def bytes(self):return self.response.contentdef text(self):return self.response.text

通过字典的方法来处理不同的解析格式，简化了 if elif 的操作，这里封装主要是将一些常用操作封装起来，让代码更简洁，当然也可以获取响应对象后，自己自由处理，最后看看封装后的使用Demo

from py_tools.connections.http import AsyncHttpClient
from py_tools.enums.http import RespFmtasync def httpx_demo(url):print("httpx_demo")async with httpx.AsyncClient() as client:resp = await client.get(url)# print(resp.text)return respasync def main():url = "https://juejin.cn/"resp_obj = await httpx_demo(url)resp_text = resp_obj.textresp_text = await AsyncHttpClient().get(url, resp_fmt=RespFmt.TEXT)if __name__ == '__main__':asyncio.run(main())

封装后简洁了许多，虽然方法有些冗余参数，但在业务中使用就不会出现好多嵌套的缩进，也牺牲了一些灵活性，因为只封装一些常用的请求操作，但一开始也想不全，只有在业务中不断的磨练，以及大家一起提建议贡献，才能慢慢的变得更好用。有时候适当的冗余封装也挺不错的。

四、同步http客户端

同步的其实 requests 已经够简洁了，没必要再封装了，这里为了统一公共库的调用，就二次封装下，思路还是跟异步的一样，有一点不一样的就是，get、post、put、delete方法返回的是 self 的引用，用于一些链式操作。一开始我想把异步的也变成链式调用，发现做不到，方法如果不await拿不到结果，返回的是 协程对象，所以一时半会弄不出来，就用了一个参数的方式来处理。

class HttpClient:"""同步HTTP客户端通过request封装，实现了常见的HTTP方法,支持设置超时时间、请求参数等，链式调用Examples:>>> HttpClient().get("http://www.baidu.com").text>>> HttpClient().get("http://www.google.com", params={"name": "hui"}).bytes>>> HttpClient().post("http://www.google.com", data={"name": "hui"}).jsonAttributes:default_timeout: 默认请求超时时间,单位秒default_headers: 默认请求头字典client: request 客户端response: 每次实例请求的响应"""def __init__(self, timeout=timedelta(seconds=10), headers: dict = None):"""构造异步HTTP客户端"""self.default_timeout = timeoutself.default_headers = headers or {}self.client = requests.session()self.response: requests.Response = Nonedef _request(self,method: HttpMethod, url: str,params: dict = None, data: dict = None,timeout: timedelta = None, **kwargs):"""内部请求实现方法创建客户端会话,构造并发送HTTP请求,返回响应对象Args:method: HttpMethod 请求方法, 'GET', 'POST' 等url: 请求URLparams: 请求查询字符串参数字典data: 请求体数据字典timeout: 超时时间,单位秒kwargs: 其他关键字参数Returns:httpx.Response: HTTP响应对象"""timeout = timeout or self.default_timeoutheaders = self.default_headers or {}self.response = self.client.request(method=method.value,url=url,params=params,data=data,headers=headers,timeout=timeout.total_seconds(),**kwargs)return self.response@propertydef json(self):return self.response.json()@propertydef bytes(self):return self.response.content@propertydef text(self):return self.response.textdef get(self, url: str, params: dict = None, timeout: timedelta = None, **kwargs):"""GET请求Args:url: 请求URLparams: 请求查询字符串参数字典timeout: 请求超时时间,单位秒Returns:self 自身对象实例"""self._request(HttpMethod.GET, url, params=params, timeout=timeout, **kwargs)return selfdef post(self, url: str, data: dict = None, timeout: timedelta = None, **kwargs):"""POST请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒Returns:self 自身对象实例"""self._request(HttpMethod.POST, url, data=data, timeout=timeout, **kwargs)return selfasync def put(self, url: str, data: dict = None, timeout: timedelta = None, **kwargs):"""PUT请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒Returns:self 自身对象实例"""self._request(HttpMethod.PUT, url, data=data, timeout=timeout, **kwargs)return selfasync def delete(self, url: str, data: dict = None, timeout: timedelta = None, **kwargs):"""DELETE请求Args:url: 请求URLdata: 请求体数据字典timeout: 请求超时时间,单位秒Returns:self 自身对象实例"""self._request(HttpMethod.DELETE, url, data=data, timeout=timeout, **kwargs)return self

五、源代码

源代码已上传到了Github，里面也有具体的使用Demo，欢迎大家一起体验、贡献。

HuiDBK/py-tools: 打造 Python 开发常用的工具，让Coding变得更简单 (github.com)

http://www.bjxfkj.com.cn/article/108196.html

相关文章：

  • 仿站工具箱ngix安装wordpress的伪静态
  • 做网站那个公司高端品牌网站建设制作需要注意什么
  • 摄影网站功能设计软件开发网站建设
  • 网站后台系统设置建筑图片大全高清
  • 网站制作对公司的作用网上装修平台哪个最好
  • 广东网站建设微信商城开发wordpress代码框
  • 建设工程有限公司 网站wordpress sharethis
  • 两学一做网站 新闻陕西建设网三类人员
  • 给自己公司做网站运营邯郸微信公众号开发
  • 网站建设分析方法新网网站空间到期停了 咋续费
  • 建设网站明细报价表龙岗区住房和建设局网站
  • 好女人生活常识网站建设购物网站商城
  • 零基础学建站重庆设计有限公司
  • 佛山网站优化公司排名网站版面在线设计
  • 开发网站的流程细节微信营销的模式
  • 开发网站需要什么技术2022甘肃交通工程建设监理有限公司网站
  • 阿里云备案网站服务内容怎么填搭建网站是要什么
  • 企业网站备案要多久网站建设店
  • 手机端网站建设郑州思茅网站建设
  • 用dw做教学网站太原网站设计
  • 网站视觉风格可在哪些网站做链接
  • 云服务器建设网站教程图片在线制作二维码生成器
  • 定制自动化营销如何seo推广
  • 自己建个购物网站建筑网络图
  • 彩票网站开发有连带万网虚拟机wordpress
  • 公司用于做网站的费用怎么做账咸宁网站建设解决方案
  • 国内有哪些响应式网站怎样做网站初中生
  • 做网站用什么软件方便购物网站每个模块主要功能
  • 如何做好公司网站建设wordpress站长
  • 网站接入协议及接入商资质对于高校类建设网站的要求