Zabbix监控数据一致性校验：解决主从节点数据不同步的问题

Yang ChenFang 2026-09-18 04:32

一、先说说我遇到的情况

做运维的朋友，手里肯定都有一两套Zabbix监控。刚开始搭建的时候，觉得一台服务器就够用了，后来发现机器越来越多，监控显得有点吃力，就想着加一台从节点，做高可用也好，做负载分担也好。主从架构一上，心里就踏实了不少。可时间一长，问题来了：主节点上看到的那台数据库服务器的内存使用率是80%，从节点上却显示成64%。明明是同一个监控项，为什么数字对不上？这要是赶上主节点宕机，切到从节点，看到的监控数据是错的，那岂不是让人抓瞎？这种情况我碰到过不止一次，后来我决定写一个自动校验的工具，专门用来揪出这种主从数据不一致的问题。

二、数据不同步的常见原因

要解决问题，先得知道它为什么会发生。Zabbix的主从同步，不是简简单单复制一下文件，它背后通常依赖数据库的主从复制。在这个过程中，任何一个小环节出错，都可能导致两边数据对不上。我把常见原因整理了一下：

第一，网络延迟。主节点已经写入的数据，可能需要几秒甚至几十秒才能传到从节点。如果你正好在同步完成前去读从节点，数据自然不一样。

第二，数据库同步中断。比如从节点的MySQL二进制日志重放出错，或者复制线程挂掉，但是Zabbix服务看起来还在运行，于是你很难发现。

第三，人为的误操作。有人直接从数据库上改了一条记录，绕过了Zabbix的正常流程，这种数据在主节点上不存在，从节点上却有。

第四，版本不一致。主节点和从节点的Zabbix版本不同，可能导致数据结构或者写入逻辑有差别。

第五，磁盘或内存异常。某一条数据写入一半就失败了，主节点上可能已经记了一条，从节点上却缺了一条。

这些原因都不容易一眼发现，靠人工比对又太累，所以我们需要一个自动化的工具。

三、校验思路：给数据做“指纹”

我的思路是：不要逐条去比较数据，而是给一段数据算出一个“指纹”。你可以把这段数据想象成一段话，指纹就是这段话的摘要。只要原数据有任何一点变化，摘要就会完全不同。这样我们只需要对比主从节点算出来的指纹，就能判断数据是否一致。

在Zabbix里，我们可以通过API从主节点读出某一个监控项在一段时间内的所有历史数据，然后把它们拼成一个大字符串。同样的方法再从从节点读一次，得到另一个字符串。最后对这两个字符串计算SHA256摘要，如果两个摘要一样，就说明两边数据一致；不一样，就说明有问题。多个监控项也可以一起处理，把它们的字符串拼在一起，算出一个总指纹，然后比较。

这种做法的好处是简单可靠，而且可以一次性覆盖很多数据。

四、用Python写一个校验脚本

这次我使用的技术栈是Python 3配合requests库。选它的原因很简单：代码简短，容易理解，而且requests库让HTTP请求变得像说话一样简单。

4.1 准备工作

先把requests库装上。如果你还没装，可以在命令行执行：

# 建议使用pip安装：
# pip install requests

装好之后，我们还需要知道主从两个Zabbix的API地址。Zabbix的API地址通常是 http://你的服务器地址/api_jsonrpc.php。还需要一个具备读取权限的账号，一般用 Admin 就够了，当然你也可以专门建一个只读账号。

4.2 登录Zabbix API并获取令牌

Zabbix的API遵循JSON-RPC协议。简单说，我们发送一段JSON格式的请求，它返回一段JSON格式的结果。登录接口的请求格式长这样：

import requests
import json

# 设置Zabbix API地址和登录信息
ZBX_URL = "http://your-zabbix-server/api_jsonrpc.php"
USERNAME = "Admin"
PASSWORD = "zabbix"

def login(url, username, password):
    """
    登录Zabbix API并返回认证令牌(token)
    """
    # 按照JSON-RPC协议构造登录请求
    payload = {
        "jsonrpc": "2.0",               # 协议版本
        "method": "user.login",         # 调用登录方法
        "params": {
            "username": username,       # 用户名
            "password": password        # 密码
        },
        "id": 1                         # 请求标识，随便填，用于对应响应
    }
    # 发送POST请求，requests会自动将dict转换成JSON
    response = requests.post(url, json=payload)
    # 解析响应
    result = response.json()
    # 如果返回结果中有error字段，说明登录失败
    if "error" in result:
        raise Exception("登录失败：" + str(result["error"]))
    # 成功的话，result["result"]就是我们的token
    return result["result"]

这段代码里，requests.post 的 json 参数会自动设置请求头为 application/json，非常方便。后面所有的API调用都需要带上这个token。

4.3 拉取某个监控项的历史数据

有了token，我们就可以查询监控项的数据。查询历史数据用的是 history.get 方法。我们需要指定监控项的ID、时间范围，以及是否要按时间排序。下面是完整的函数：

def fetch_item_values(url, token, item_id, since, until):
    """
    从Zabbix API拉取指定监控项在时间范围内的历史数据
    """
    # 构造请求体
    payload = {
        "jsonrpc": "2.0",
        "method": "history.get",        # 方法名
        "params": {
            "itemids": item_id,         # 监控项ID
            "time_from": since,         # 起始时间戳
            "time_till": until,         # 结束时间戳
            "output": "extend",         # 返回所有字段
            "sortfield": "clock",       # 按时间排序
            "sortorder": "ASC"          # 升序，便于拼接
        },
        "auth": token,                  # 刚才登录拿到的令牌
        "id": 2                         # 请求标识换成2
    }
    # 发送请求
    response = requests.post(url, json=payload)
    result = response.json()
    # 检查错误
    if "error" in result:
        raise Exception("查询失败：" + str(result["error"]))
    # 把每条记录的clock(时间戳)和value(值)拼成一个字符串
    data_str = ""
    for record in result["result"]:
        data_str += f"{record['clock']}:{record['value']};"
    return data_str

拼成的字符串大概长这样：1612345678:80.5;1612345738:80.6;...。这个字符串已经包含了所有需要校验的信息。

4.4 计算数据指纹

拿到字符串之后，我们用Python标准库中的hashlib来计算SHA256摘要：

import hashlib

def calc_fingerprint(data_str):
    """
    计算一段数据的SHA256指纹
    """
    # 创建sha256哈希对象
    sha = hashlib.sha256()
    # 更新内容，字符串需要转成utf-8字节
    sha.update(data_str.encode("utf-8"))
    # 返回十六进制格式的摘要字符串
    return sha.hexdigest()

注意，我们故意不直接用MD5，是因为SHA256在碰撞概率上更安全。虽然单看性能比MD5慢一点点，但是对监控数据来说完全够用。

4.5 比较主从节点的指纹

现在我们可以把前面的函数组合起来，写一个主流程。假设主节点API地址是 http://master.example.com/api_jsonrpc.php，从节点是 http://slave.example.com/api_jsonrpc.php。我们校验某个监控项最近1小时的数据：

import time

def main():
    # 主节点信息
    master_url = "http://master.example.com/api_jsonrpc.php"
    master_token = login(master_url, "Admin", "zabbix")

    # 从节点信息
    slave_url = "http://slave.example.com/api_jsonrpc.php"
    slave_token = login(slave_url, "Admin", "zabbix")

    # 要校验的监控项ID，实际使用时请替换成你环境里的ID
    item_id = "29174"
    # 获取当前时间戳，作为“直到”时间
    time_till = int(time.time())
    # 往前推1小时，作为“从”时间
    time_from = time_till - 3600

    # 分别从主从节点拉取数据
    master_data = fetch_item_values(master_url, master_token, item_id, time_from, time_till)
    slave_data = fetch_item_values(slave_url, slave_token, item_id, time_from, time_till)

    # 计算指纹
    master_fp = calc_fingerprint(master_data)
    slave_fp = calc_fingerprint(slave_data)

    # 对比指纹，输出结论
    if master_fp == slave_fp:
        print("校验通过：主从节点数据一致")
    else:
        print("校验失败：主从节点数据不一致")
        print("主节点指纹：" + master_fp)
        print("从节点指纹：" + slave_fp)

# 运行主程序
if __name__ == "__main__":
    main()

就这么简单，一个完整的主从数据一致性校验脚本就出来了。你只要把它放到服务器上定时执行，就能每天自动检查。

4.6 批量校验一个主机下的所有监控项

实际监控环境里，一台主机下面可能有几十个监控项。如果每个都手工指定ID，那太浪费时间。我们可以改进一下，通过 item.get 方法获取主机下所有监控项，然后循环校验。下面这个函数可以拿到一个主机下所有的监控项：

def get_all_items(url, token, host_id):
    """
    获取指定主机下的所有监控项ID和名称
    """
    payload = {
        "jsonrpc": "2.0",
        "method": "item.get",
        "params": {
            "hostids": host_id,
            "output": ["itemid", "name"]
        },
        "auth": token,
        "id": 3
    }
    response = requests.post(url, json=payload)
    result = response.json()
    if "error" in result:
        raise Exception("获取监控项失败：" + str(result["error"]))
    # 返回所有监控项的列表，每个元素有itemid和name
    return result["result"]

接着，我们写一个函数，把一个主机下所有监控项的数据合并成一个总数据，再算总指纹：

def build_host_fingerprint(url, token, host_id, since, until):
    """
    计算一个主机下所有监控项合并后的总指纹
    """
    items = get_all_items(url, token, host_id)
    total_data = ""
    # 循环每一个监控项
    for it in items:
        item_id = it["itemid"]
        # 拉取该监控项的数据
        data_str = fetch_item_values(url, token, item_id, since, until)
        # 拼接到总数据字符串中
        total_data += data_str
    # 对总数据计算指纹
    return calc_fingerprint(total_data)

然后在主流程里，只需要把单独监控项的部分替换成下面这些代码：

# 假设我们要比对主机ID是10105
host_id = "10105"
master_fp = build_host_fingerprint(master_url, master_token, host_id, time_from, time_till)
slave_fp = build_host_fingerprint(slave_url, slave_token, host_id, time_from, time_till)

这样一次就能覆盖整台主机的所有监控项。

4.7 额外手段：通过Python检查数据库同步状态

如果上面的指纹比对发现不一致，那么问题很有可能出在数据库同步这一层。作为关联技术，我再介绍一下怎么用Python去读取从库的同步状态。假设你要检查的是一个MySQL从库，你可以用Python的 subprocess 调用 mysql 命令，把 SHOW SLAVE STATUS 的输出抓回来，然后解析出关键字段：

import subprocess

def check_mysql_slave_status():
    """
    通过mysql客户端获取从库同步状态（需要mysql client已安装且可访问）
    """
    # 这里为了安全，建议使用配置文件或环境变量传用户名和密码
    cmd = "mysql -uzabbix_sync -psecret -e 'SHOW SLAVE STATUS\\G'"
    # 执行命令并把输出保存
    output = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT)
    # 转成字符串返回
    return output.decode("utf-8")

拿到输出之后，你可以搜索 Seconds_Behind_Master 这个字段，如果它不是0，就说明有延迟。这只是一个切入点，真正要修复数据库同步问题，还需要结合数据库日志来进一步定位。

4.8 把校验结果推送到你的手机

脚本跑出来了，但还是需要人看。我们可以顺便加一个“通知”功能，把校验结果发送到飞书或者钉钉。这里以钉钉机器人举例：

def send_dingtalk(message):
    """
    通过钉钉机器人发送消息
    """
    # 这里是钉钉机器人的Webhook地址，请替换成你自己的
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
    payload = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    requests.post(webhook_url, json=payload)

然后在比较指纹的代码里，如果发现不一致，就调用这个方法：

if master_fp != slave_fp:
    send_dingtalk("警告：Zabbix主从数据不一致，请检查！")

这样，即使你不在电脑前，也能第一时间收到通知。

五、适用场景

说了这么多，这个方案在哪些场合最有用？我认为至少有这几种：

第一，日常巡检。每一个小时或者每天凌晨跑一次，把校验脚本的结果写入日志，如果发现不一致就告警。

第二，主备切换前验证。在做Zabbix高可用切换之前，先跑一遍脚本，确保两边数据几乎是一致的，再决定要不要切。

第三，同步故障恢复后验证。当你修复了数据库同步问题之后，用这个脚本来确认一切恢复正常。

第四，跨机房监控数据核对。主从节点分别在不同的地方，网络链路更长，更有可能出现数据不一致，用自动校验可以及时发现问题。

六、技术优缺点

6.1 优点

第一个优点就是省心。脚本全自动，不用人工盯着两台服务器看。

第二个优点是覆盖面广。只要API能查到的历史数据，都可以校验，不限于某一类监控项。

第三个优点是逻辑简单。指纹对比非常直观，代码量小，出了问题也容易排查。

6.2 缺点

它也有不足。首先，它只能发现不一致，不能定位到具体是哪一条记录出了错。想要知道差异细节，还需要进一步拉取数据做逐条对比。

其次，频繁调用API会给Zabbix服务增加额外的压力，尤其是在监控项很多的时候。所以脚本的运行频率不要太高。

最后，如果主从节点所在服务器的时间不一致，我们的时间窗口就会错开，这会导致误报。所以使用前要先保证系统时间同步。

七、注意事项

为了让你把这个工具用好，我再多说几个容易踩坑的地方。

一是别拿最新那一刻的数据来比对。因为同步延迟是客观存在的，你10点整去查，可能主节点还有最后一条数据没同步到从节点。最好的做法是对比一段时间之前的数据，比如10分钟前，这样两边都已经稳定了。

二是API账号的权限要够。如果你的账号只给了读主机配置的权限，没有读history的权限，那么 history.get 会报错。建议使用Zabbix管理员账号，或者创建一个带 Zabbix Historian 权限的用户。

三是请求频率要控制。如果一台主机有上千个监控项，你每秒拉一次，那肯定会对数据库造成压力。可以把脚本设置成在业务低谷期跑，并且给每次请求加一点延时，比如 time.sleep(0.1)。

四是注意监控项类型。有的监控项是文本型，有的可能是浮点型，拼接字符串的方式都一样。但要注意，如果数据中存在分号或冒号，可能会破坏字符串格式。不过Zabbix的value一般不会带这些符号，所以我们用分号分隔是安全的。

五是最后一点，指纹一致不等于数据绝对正确。假如主从节点都因为同一原因丢失了同一条数据，那么两边指纹还是相同。所以这个工具的核心价值是“发现不同步”，而不是“保证绝对准确”。

八、总结

Zabbix主从节点数据不同步，是一个既常见又让人头疼的问题。通过Python脚本调用Zabbix API，拉取历史数据，再用SHA256指纹做主从对比，我们可以用非常低的成本实现自动化校验。从单个监控项到整台主机，再到定时任务和消息通知，整个链路都很清晰。

在定位不同步问题时，这个脚本能帮你快速缩小范围，之后再结合数据库同步状态、网络日志等信息，就能找到真正的根因。希望这些内容能对你有所帮助，让你不再为监控数据对不上而烦恼。

围绕“Zabbix监控数据一致性校验：解决主从节点数据不同步的问题”参与讨论

0 条

Zabbix监控数据一致性校验：解决主从节点数据不同步的问题

一、先说说我遇到的情况

二、数据不同步的常见原因

三、校验思路：给数据做“指纹”

四、用Python写一个校验脚本

4.1 准备工作

4.2 登录Zabbix API并获取令牌

4.3 拉取某个监控项的历史数据

4.4 计算数据指纹

4.5 比较主从节点的指纹

4.6 批量校验一个主机下的所有监控项

4.7 额外手段：通过Python检查数据库同步状态

4.8 把校验结果推送到你的手机

五、适用场景

六、技术优缺点

6.1 优点

6.2 缺点

七、注意事项

八、总结

评论

关联文章

AI助手

一、先说说我遇到的情况

二、数据不同步的常见原因

三、校验思路：给数据做“指纹”

四、用Python写一个校验脚本

4.1 准备工作

4.2 登录Zabbix API并获取令牌

4.3 拉取某个监控项的历史数据

4.4 计算数据指纹

4.5 比较主从节点的指纹

4.6 批量校验一个主机下的所有监控项

4.7 额外手段：通过Python检查数据库同步状态

4.8 把校验结果推送到你的手机

五、适用场景

六、技术优缺点

6.1 优点

6.2 缺点

七、注意事项

八、总结

同分类面试题

关联文章

AI助手