☰
Python批量域名转IP:从socket到dnspython的并发解析实践
2026/10/10 18:35:36 网站建设 项目流程

上周帮朋友处理一批安全日志,日志里抽出来的域名少说也有三四百条。当时第一反应是:不能一个个去点网页工具查。解析域名这种重复劳动,最适合丢给Python做批处理。事后证明这个决定是对的——从拿到列表到输出一份干净的《域名-IP对照表》,前后也就几分钟。这篇文章就围绕"用Python批量域名转换IP"这件事,把我踩过的坑、最终采用的方案、以及值得留意的细节都记录下来。不管你是要处理几百条告警日志的运维,还是需要做资产梳理、写自动化策略的工程师,都能直接把这里面的思路套到自己的场景里。

1. 解析需求先拆开:三种场景和一套通用逻辑

1.1 什么样的任务才需要批量解析

批量域名转IP听起来是个小工具,但它解决的是一个很典型的工程问题:把散落的数据变成可计算的指标。

最常见的三类场景如下:

  • 安全日志关联与资产梳理:告警平台里的日志字段经常是域名,比如连接了某个可疑域名、访问了某个外部站点。要把这些域名和IP情报库做碰撞,就必须先把域名批量解析成IP。只查一两个域名可以用在线工具,一旦上了几十上百个,手点就不现实了。
  • 防火墙/访问控制策略生成:一些防火墙策略只支持填写IP网段,不支持域名。运维拿到一批业务域名要做出口白名单,就得批量转成IP,然后按C段聚合,再写成策略组。
  • 监控与调度观察:业务域名解析到的IP经常变动,比如CDN调度、负载均衡调整、故障切换。定期跑一遍批量解析,对比前后两次结果,就能快速发现解析漂移,判断业务是不是被切到了别的节点。

这类任务看起来零散,但底层逻辑很一致:输入一批域名,输出一批IP,中间还要处理解析失败、多IP、顺序变化这些边界情况。把这个逻辑封装成一个脚本,能反复用,才是批量转IP的最大价值。

1.2 输入输出长什么样:先定规约再写代码

写脚本之前,我强烈建议先想清楚输入和输出格式。大多数人直接开个文件每一行写个域名就算了,但等到真要复用时,格式问题会反咬一口。

我的建议是输入文件用最朴素的纯文本,一行一个域名,允许空行和#开头注释。编码统一用UTF-8,避免Windows记事本另存为GBK导致的中文乱码——虽然域名理论上全是ASCII,但注释里很可能写中文。域名一律小写,去掉末尾的点(FQDN形式),不去管http://或https://前缀,那种脏数据应该在预处理阶段清洗掉,而不是让解析函数背着。

输出文件我推荐CSV而不是TXT拼JSON。CSV一行是一条完整记录,列分别是:域名、解析到的IP列表、错误码。IP之间用空格分隔,错误码留空表示成功。这样后续无论用Excel筛选、用pandas做统计,还是直接写awk处理,都很顺手。

1.3 需求边界:哪些情况必须提前想好

一个看似简单的"域名换IP"实际上包含了好几个必须明确回答的问题:

  • 域名解析不到时,是跳过、标记还是重试?
  • 一个域名解析出多个IP,是只保留第一个还是全部保留?
  • 重复域名要不要去重?
  • 数十条、数百条还是数万条,性能要求完全不同?
  • 这台机器所在的网络环境解析是否可信?

这些边界条件不提前定义好,脚本写出来就只能在你自己的机器、你自己那一批数据上跑一次。真正好用的批量解析脚本,应该是换一份域名列表、换一台机器也能稳定跑完的类型。

2. 先解决选型:socket和dnspython到底差在哪

2.1 socket自带解析函数的真相

很多人的第一反应是用Python自带的socket模块,毕竟它零依赖。socket.gethostbyname('example.com')确实能返回一个IP,但如果你准备用它跑批量任务,我劝你先冷静。

socket.gethostbyname有四个明显问题。

第一,它只返回一个IP。现代互联网里一个域名对应多个A记录太常见了,CDN、负载均衡、多活机房都会让域名返回多个IP。只拿第一个,结果可能是不完整甚至错误的。

第二,它全权交给操作系统的解析流程,也就是读/etc/resolv.conf或Windows的DNS设置。你在本机跑出来的结果,和运维在DNS服务器上看到的结果可能根本对不上。批量任务需要的是可控、可复现的解析,而不是"随系统心情"的解析。

第三,它不好设置超时。默认情况下,如果DNS服务器不响应,应用层可能卡在那里很久。虽然socket内部有底层超时机制,但通过gethostbyname这一层你几乎没法精细控制。

第四,它对错误原因基本不区分。域名不存在、DNS服务器挂了、网络不可达,它给你的都是同一个gaierror类异常,连个错误码都要费劲解析。

还有一个容易被忽略的细节:gethostbyname在某些系统上会受/etc/hosts文件影响。如果你本机hosts里写了某个测试域名的映射,解析结果根本就不是真实DNS的答案,批量结果会被"污染"而不自知。

2.2 为什么我要用dnspython

批量解析场景下,我最终选择的是dnspython。它本质上是一个用纯Python实现的完整DNS客户端,不走系统解析器,而是自己构造DNS报文、自己管理超时、自己解析响应。

它有几个对批处理特别有价值的能力:

  • 可以显式指定nameserver,例如用223.5.5.5或8.8.8.8,结果一致性有保障;
  • 可以设置timeout和lifetime,从根上避免单条查询卡死整个任务;
  • 异常体系完整,NXDOMAIN、NoAnswer、Timeout、NoNameservers分得很清楚;
  • 返回的rrset里可以拿到全部A记录,而不是只给一个。

安装方式很简单:

pip install dnspython

2.3 先把单个域名的解析跑通

批量之前,一定要先把单条查询调通。我手头最常用的单条解析片段是这样:

import dns.resolver resolver = dns.resolver.Resolver() resolver.nameservers = ['223.5.5.5', '119.29.29.29', '8.8.8.8'] resolver.timeout = 3.0 resolver.lifetime = 5.0 answers = resolver.resolve('example.com', 'A') for rdata in answers: print(rdata.address)

这段代码有几个细节值得解释。

nameservers这里我放了三个公共DNS。为什么不只用本机配置的DNS?因为批量解析过程中,本机DNS如果正好是某个内网转发器,很可能有缓存策略或污染问题,导致解析结果异常。显式指定多个公共DNS,至少能让结果可对比、可复现。

timeout和lifetime的区别很多人搞错。timeout控制的是单次网络读写的超时,lifetime控制的是整个查询流程允许消耗的最大时间,包括重试、等待多个服务器响应的时间。批量场景里lifetime我一般设成5.0秒,如果DNS服务器持续无响应,超过这个时间就直接判定超时,不让单条数据拖死整个队列。

resolver.resolve('example.com', 'A')会自动跟随CNAME链,最终给你返回A记录集合。如果你要保留下游所有IP,直接遍历answers即可。

3. 并发批量解析:从串行改造到线程池

3.1 串行解析到底慢在哪

如果把上面的代码套一个for循环,几百个域名确实能跑,但你大概率会被速度气死。DNS查询是典型的I/O密集型操作,大部分时间都耗在等待网络响应的往返延迟上。一个域名如果运气不好,要等一两秒才返回,一百个域名串行跑就是一百多秒。

这时候就该上并发。并发不是炫技,而是对"耗时不均"这种常态的妥协:有的域名50毫秒返回,有的要卡到接近超时才回来,并发能把这些等待时间重叠起来。

3.2 一份可复用的批量解析脚本

下面这个版本是我在实际任务里用过的,结构比较完整,可以直接存成.py文件跑。输入文件domains.txt一行一个域名,输出文件result.csv,每一行是域名、IP列表、错误码三列。

import csv from concurrent.futures import ThreadPoolExecutor, as_completed import dns.resolver INPUT_FILE = "domains.txt" OUTPUT_FILE = "result.csv" NAMESERVERS = ["223.5.5.5", "119.29.29.29", "8.8.8.8"] MAX_WORKERS = 50 TIMEOUT = 3.0 LIFETIME = 5.0 def create_resolver(): r = dns.resolver.Resolver() r.nameservers = NAMESERVERS r.timeout = TIMEOUT r.lifetime = LIFETIME return r def resolve_domain(domain): domain = domain.strip().lower().rstrip(".") if not domain: return domain, [], "empty" resolver = create_resolver() try: answers = resolver.resolve(domain, "A") ips = sorted({rdata.address for rdata in answers}) return domain, ips, "" except dns.resolver.NXDOMAIN: return domain, [], "NXDOMAIN" except dns.resolver.NoAnswer: return domain, [], "NoAnswer" except dns.resolver.NoNameservers: return domain, [], "NoNameservers" except dns.exception.Timeout: return domain, [], "Timeout" except Exception as exc: return domain, [], type(exc).__name__ def load_domains(path): domains = [] with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip().lower().rstrip(".") if line and not line.startswith("#"): domains.append(line) return domains def main(): domains = load_domains(INPUT_FILE) print(f"待解析域名: {len(domains)}") with open(OUTPUT_FILE, "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(["domain", "ip_list", "error"]) with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = {executor.submit(resolve_domain, d): d for d in domains} for i, future in enumerate(as_completed(futures), 1): domain, ips, err = future.result() writer.writerow([domain, " ".join(ips), err]) if i % 50 == 0: print(f"进度: {i}/{len(domains)}") print(f"完成,结果写入 {OUTPUT_FILE}") if __name__ == "__main__": main()

这段代码里的几个选择,都是我实际踩过坑之后定下来的。

每个worker函数里都新建一个Resolver()实例,而不是全局共用一个。dnspython的Resolver对象内部有些状态字段,多线程共享容易出现不可预期的问题。反正创建Resolver很轻,每次查询都新建,省心。

IP列表用sorted({rdata.address for rdata in answers}),先通过集合去重,再排序。这样同一个域名多次解析出来的结果,不会因为乱序导致对比时误报"解析变化"。

错误处理兜底了一个except Exception,把所有没预料到的异常类型名称写进错误列。这是排障时最实用的设计——你不会想看到脚本突然崩掉,然后对着堆栈猜是哪条域名出了问题。

3.3 并发参数怎么定:池大小、超时与重试

MAX_WORKERS我推荐50。为什么不设500?因为DNS请求最终要打到上游DNS服务器上,公共DNS对单源IP的并发请求是有限流策略的,50个并发已经能让普通的千条级任务在一两分钟内完成,再高只会引发超时和限流,得不偿失。

超时参数需要根据网络环境微调。如果你跑脚本的机器在IDC机房,网络质量好,timeout可以缩到2秒,lifetime缩到4秒。如果是在普通办公网络里跑,公共DNS有时响应很慢,3秒和5秒是更稳妥的值。

关于重试,我的做法是:第一轮跑完后,单独把错误列为Timeout的域名再跑一遍。因为批量并发时,瞬时大量请求涌向DNS服务器,部分查询超时属于正常的排队现象,不是域名真的有问题。第二轮串行查这些遗漏项,配上time.sleep(0.5)的间隔,成功率会高很多。代码不必写得多复杂,最简单的方式就是从CSV里筛出error列为Timeout的行,丢回resolve_domain再执行一次。

4. 输出落地和异常处理:脚本好用才叫完事

4.1 CSV落盘要放进主线程,别在子线程里写文件

我发现很多初学者会在每个worker线程里自己打开文件,写完再关。这个做法在批量任务里会埋雷:多个线程同时写同一个文件句柄,轻则内容交错,重则文件损坏。

上面的脚本里,我让worker只负责返回(domain, ips, err)元组,由主线程统一用CSV writer写入文件。ThreadPoolExecutor配合as_completed可以做到"完成一条写一条",内存占用小,文件写入顺序也完全可控。

真正跑起来后,你还会发现一个细节:CSV文件的编码要统一指定为utf-8,并且newline=""不能丢。这两个参数配合不好,在Windows上写出来的CSV可能一行变两行,或者中文注释乱码。

4.2 解析失败的归类:不要把错误笼统扔进一个except

批量解析最大的坑不是解析本身,而是失败原因被混淆。一个域名查不到IP,可能是因为域名真不存在,也可能因为它只有AAAA记录而没有A记录,也可能纯粹是DNS服务器超时。这三种情况对后续处理的意义完全不同。

我的脚本里把异常拆成了几类:

错误类型含义常见处理方式
NXDOMAIN域名不存在标记后人工复核,可能是拼写错误或域名已注销
NoAnswer域名存在但没有A记录检查是否需要查询AAAA记录,或对方只配置了邮件相关记录
Timeout查询超时第一轮结束后单独重试,多半能查到
NoNameservers上游DNS服务器异常换一组nameserver重试
其他异常未预料的错误把异常类名写入错误列,后续针对性排查

这样分类之后,CSV里的error列就能直接当筛选条件用。如果在防火墙策略场景里,NXDOMAIN的域名可以直接从白名单删除;Timeout的则等重试结果出来再决定。

4.3 长任务运行中的进度与断点续跑

域名数量到了几千条,脚本运行时间可能达到几分钟。这时候有两个体验问题必须处理:一是进度反馈,二是挂了之后重新跑的成本。

进度方面,每处理50条打印一次进度就够了,不需要每一条都print。频繁的标准输出在大量并发下反而会拖慢运行,而且刷屏之后根本没法看历史信息。我习惯再加一个参数,把没跑完的域名单独写一个pending.txt,这样即使进程中断,下次只需要对残留文件重新执行,不用从头再来。

断点续跑的实现思路很简单:先读现有的结果文件,把已成功解析的域名放入一个集合,读输入域名时跳过这个集合里的域名。这样重复运行脚本不会重复请求DNS,不会惹恼上游服务器。

5. 进阶优化:多IP、缓存与更稳定的解析源

5.1 一个域名返回多个A记录,别只拿第一个

前面提到过,生产环境里一个域名对应多个IP是常态。CDN节点、多机房负载均衡都会让同一个域名在DNS层面返回多条A记录。

实际演练时,我曾经因为只取第一个IP,导致防火墙白名单漏掉了一个备用节点。业务一发生切流,对应区域直接断开。所以最终脚本里必须用answers.rrset或直接遍历answers来拿全部IP。

想查看一条记录的TTL也很简单:

answers = resolver.resolve(domain, "A") if answers.rrset is not None: ttl = answers.rrset.ttl print(domain, ttl)

TTL值在批量解析里看起来不重要,但如果你的任务是持续监控域名解析变化,它就是你设计缓存时间的依据。

5.2 CNAME链和PTR记录要不要处理

批量转IP这个场景下,CNAME一般不需要专门处理。dns.resolver.resolve(domain, "A")会自动把CNAME链跟到底,最终返回的是真实的目标IP。你只需要明白一点:结果里出现的IP可能是CDN厂商的节点,而不是你业务服务器本身。

反过来,有些人会想把IP反查成域名做进一步分析。这个用dns.reversename.from_address()构造PTR查询即可,但PTR记录经常没有完整配置,尤其在云上,反查成功率并不高。我在实际任务里很少用PTR做自动判断,它更适合作为人工辅助手段。

5.3 解析源策略:多个nameserver怎么配合

批量任务的解析源策略会直接影响结果的一致性。我常用的组合是223.5.5.5(阿里)、119.29.29.29(腾讯)、8.8.8.8(Google)。前面两个在国内网络环境普遍响应快,8.8.8.8作为兜底用于结果比对。

这里要给一个警示:不同DNS服务器对同一域名的解析结果可能有差异,这很正常。当批量结果出现"诡异IP"时,不要急着下结论,先把同一域名用其他DNS各查一遍,再结合ttl判断是不是历史缓存残留。

如果任务量特别大,还可以从系统配置的DNS和公共DNS之间做轮询,避免单一DNS被限流。轮询不需要复杂的算法,维护一个nameserver列表和一个全局计数器,每次创建Resolver时用nameservers[index % len(nameservers)]选一个就行。

5.4 TTL缓存:长时间运行的脚本必须考虑

如果你的任务不是跑一次就结束,而是每隔几分钟批量解析一遍,那必须考虑TTL缓存。

最简单的做法是在内存里放一个字典,key是域名,value是(ip列表, 缓存到期时间)。每次查域名之前先看缓存是否有效,有效就直接返回,无效再发DNS请求。TTL就是缓存时间上限,可以把rrset.ttl作为过期时间来源。

这个缓存的收益在大批量监控场景里非常明显。几十个域名看起来无所谓,但当域名数量上千、监控频率又高时,不加缓存会导致DNS查询量暴增,很快触发限流。

5.5 输出格式扩展:从CSV到JSON和Excel

CSV是通用格式,但有些场景需要更友好的输出。

如果下游是自动化平台,JSON更合适。解析结果转JSON不难:读CSV的每一行,组成{domain: {"ips": [...], "error": ""}}结构,用json.dump写出去即可。如果下游是给业务同事看的,Excel更直观,可以用pandas的read_csv读入CSV再to_excel导出,也可以直接用openpyxl逐行写。但说实话,pandas一页页面能解决的事,不值得在核心脚本里引入额外依赖。

好多时候,批量解析脚本的价值不在解析本身,而在后续处理的效率。只要结果文件里字段清晰、错误可控,下游用什么工具加工都只是顺手的事。

最后再分享一个小经验:写这类脚本时,不要把输入文件路径和nameserver列表写死在代码里,用argparse或至少常量来暴露它们。你会发现自己三个月后再用这个脚本时,最感谢的就是当初留出来的那三个命令行参数。一个工具能不能"越用越顺手",往往取决于你把多少细节留给了未来的自己。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询