You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python Locust如何指定自定义DNS服务器进行URL解析?

Locust指定自定义DNS服务器实现方法

可行,但Locust的HttpUser.client本身没有内置dns参数,需要通过手动解析域名+设置Host头的方式实现,效果和nslookup google.com 8.8.8.8类似。

实现思路

Locust的HTTP客户端基于requests库,而requests默认使用系统DNS解析域名。要指定自定义DNS,需要:

  1. 用DNS解析库手动查询指定DNS服务器获取目标域名的IP
  2. 用解析出的IP替换请求URL中的域名
  3. 设置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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.20 22:06:33