Python Locust如何指定自定义DNS服务器进行URL解析?
Locust指定自定义DNS服务器实现方法
可行,但Locust的HttpUser.client本身没有内置dns参数,需要通过手动解析域名+设置Host头的方式实现,效果和nslookup google.com 8.8.8.8类似。
实现思路
Locust的HTTP客户端基于requests库,而requests默认使用系统DNS解析域名。要指定自定义DNS,需要:
- 用DNS解析库手动查询指定DNS服务器获取目标域名的IP
- 用解析出的IP替换请求URL中的域名
- 设置
Host请求头,确保服务器能正确识别目标站点
具体实现代码
1. 依赖安装
首先安装dnspython库用于DNS解析:
pip install dnspython
2. 自定义请求方法(推荐)
封装一个支持指定DNS的请求方法,让用法更接近你期望的形式:
from locust import HttpUser, task import dns.resolver from urllib.parse import urlparse class CustomHttpUser(HttpUser): def request_with_dns(self, method, path, dns_server, **kwargs): # 从base_url提取目标域名 parsed_base = urlparse(self.client.base_url) target_domain = parsed_base.netloc # 用指定DNS服务器解析域名 resolver = dns.resolver.Resolver() resolver.nameservers = [dns_server] ip_record = resolver.resolve(target_domain, "A")[0].address # 构造带IP的请求URL target_url = f"{parsed_base.scheme}://{ip_record}{path}" # 强制设置Host头,避免服务器识别错误 headers = kwargs.pop("headers", {}) headers["Host"] = target_domain kwargs["headers"] = headers # 发起请求 return self.client.request(method, target_url, **kwargs) # 封装get快捷方法 def get_with_dns(self, path, dns_server, **kwargs): return self.request_with_dns("GET", path, dns_server, **kwargs) class User(CustomHttpUser): base_url = "https://your-target-domain.com" token = "your-auth-token" @task def get_data(self): headers = { 'authToken': self.token, 'Content-Type': 'application/json' } # 调用自定义方法,传入指定DNS服务器 res = self.get_with_dns("/data", dns_server="8.8.8.8", headers=headers)
3. 简化版(单次解析)
如果只需要固定DNS服务器,也可以在用户初始化时提前解析IP:
from locust import HttpUser, task import dns.resolver def resolve_domain(domain, dns_server): resolver = dns.resolver.Resolver() resolver.nameservers = [dns_server] return resolver.resolve(domain, "A")[0].address class User(HttpUser): target_domain = "your-target-domain.com" custom_dns = "8.8.8.8" token = "your-auth-token" def on_start(self): # 初始化时解析IP,缓存结果 self.target_ip = resolve_domain(self.target_domain, self.custom_dns) @task def get_data(self): headers = { 'authToken': self.token, 'Content-Type': 'application/json', 'Host': self.target_domain } # 直接用IP发起请求 res = self.client.get(f"http://{self.target_ip}/data", headers=headers)
注意事项
- SSL验证:如果是HTTPS请求,部分服务器可能会因为IP与证书域名不匹配报错,此时可以添加
verify=False参数关闭验证(仅测试环境使用),或确保客户端支持SNI(大多数现代Python环境默认支持)。 - IP缓存:频繁解析DNS会影响性能,建议在
on_start中解析一次并缓存IP,若域名IP经常变动,可定时重新解析。 - 多IP处理:如果域名对应多个IP,可随机选择一个模拟真实请求分布。
内容的提问来源于stack exchange,提问作者alex
相关产品推荐
相关产品推荐

