保姆级教程：手把手教你用企业微信机器人搞定Zabbix 6.0告警（附脚本和避坑点）
企业微信机器人无缝对接Zabbix告警系统实战指南
开篇：为什么选择企业微信机器人作为Zabbix告警通道？
在运维监控领域，告警信息的及时触达往往决定着故障响应速度。传统邮件和短信告警存在延迟高、成本昂贵的问题，而企业微信机器人凭借其即时性、零成本和易集成特性，正成为越来越多企业的首选方案。本文将带您从零开始，完成Zabbix与企业微信机器人的深度整合，特别针对Zabbix 6.0版本（兼容5.x）提供经过实战检验的配置方案。
1. 企业微信机器人创建与配置
1.1 创建企业微信机器人
- 登录企业微信管理后台（需企业管理员权限）
- 进入「应用管理」→「自建应用」→「创建应用」
- 填写应用名称（如"Zabbix告警"）、选择可见范围
- 创建完成后，记录以下关键信息：
- AgentId：应用详情页可见
- CorpId：企业信息页面获取
- Secret：应用详情页的"Secret"项
注意：Secret仅在创建时显示一次，请务必妥善保存。若遗失需重新生成。
1.2 配置机器人Webhook
# 获取access_token示例（需替换实际参数） curl -G "https://qyapi.weixin.qq.com/cgi-bin/gettoken" \ --data-urlencode "corpid=YOUR_CORPID" \ --data-urlencode "corpsecret=YOUR_SECRET"获取到的access_token有效期为2小时，需定时刷新。建议通过以下Python脚本自动管理：
import requests import time class WeComToken: def __init__(self, corpid, corpsecret): self.corpid = corpid self.corpsecret = corpsecret self._token = None self._expires_at = 0 def get_token(self): if time.time() < self._expires_at: return self._token url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corpid}&corpsecret={self.corpsecret}" resp = requests.get(url).json() self._token = resp['access_token'] self._expires_at = time.time() + resp['expires_in'] - 300 # 提前5分钟刷新 return self._token2. Zabbix告警脚本开发
2.1 创建告警脚本文件
在Zabbix服务器上创建脚本存放目录：
mkdir -p /usr/lib/zabbix/alertscripts chown zabbix:zabbix /usr/lib/zabbix/alertscripts chmod 755 /usr/lib/zabbix/alertscripts创建Python告警脚本wecom_alert.py：
#!/usr/bin/env python3 import json import requests import sys def send_wecom_message(corp_id, secret, agent_id, to_user, content): # 获取access_token token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corp_id}&corpsecret={secret}" token_resp = requests.get(token_url).json() if token_resp.get('errcode') != 0: raise Exception(f"获取token失败: {token_resp}") # 发送消息 send_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token_resp['access_token']}" payload = { "touser": to_user, "msgtype": "markdown", "agentid": agent_id, "markdown": { "content": content }, "safe": 0 } send_resp = requests.post(send_url, json=payload).json() return send_resp if __name__ == "__main__": try: # 从命令行参数读取配置 corp_id = sys.argv[1] secret = sys.argv[2] agent_id = sys.argv[3] to_user = sys.argv[4] subject = sys.argv[5] message = sys.argv[6] # 构造消息内容 content = f"**{subject}**\n\n{message}" # 发送消息 result = send_wecom_message(corp_id, secret, agent_id, to_user, content) if result.get('errcode') == 0: print("消息发送成功") sys.exit(0) else: print(f"发送失败: {result}") sys.exit(1) except Exception as e: print(f"执行出错: {str(e)}") sys.exit(1)2.2 脚本权限与测试
设置脚本权限并测试：
chmod 755 /usr/lib/zabbix/alertscripts/wecom_alert.py chown zabbix:zabbix /usr/lib/zabbix/alertscripts/wecom_alert.py # 测试脚本 /usr/lib/zabbix/alertscripts/wecom_alert.py \ $CORP_ID $SECRET $AGENT_ID "@all" \ "测试标题" "这是一条测试消息内容"3. Zabbix控制台配置
3.1 创建告警媒介类型
- 登录Zabbix控制台 → 管理 → 告警媒介类型 → 创建媒介类型
- 配置参数：
- 名称：企业微信告警
- 类型：脚本
- 脚本名称：wecom_alert.py
- 脚本参数配置（按顺序）：
{ALERT.SENDTO} {ALERT.SUBJECT} {ALERT.MESSAGE}
3.2 配置告警动作
- 创建新的动作：配置 → 动作 → 创建动作
- 设置动作条件（示例）：
- 触发器 = 你的触发器名称
- 触发器严重性 ≥ 一般严重性
- 操作配置：
- 默认操作步骤持续时间：1h
- 默认标题：
【{TRIGGER.STATUS}】{TRIGGER.NAME} - 默认消息：
**主机**: {HOST.NAME} ({HOST.IP}) **问题**: {TRIGGER.NAME} **严重性**: {TRIGGER.SEVERITY} **时间**: {EVENT.DATE} {EVENT.TIME} **当前值**: {ITEM.VALUE} **事件ID**: {EVENT.ID}
3.3 用户告警媒介配置
- 进入管理 → 用户 → 选择相应用户 → 告警媒介
- 添加新的媒介：
- 类型：选择刚创建的企业微信告警
- 收件人：填写企业微信用户ID（多个用|分隔），或@all通知所有人
- 重要程度：根据需求选择
4. 高级配置与故障排查
4.1 消息模板优化
建议使用Markdown格式增强可读性：
**[{TRIGGER.STATUS}] {TRIGGER.NAME}** > **主机**: {HOST.NAME} ({HOST.IP}) > **时间**: {EVENT.DATE} {EVENT.TIME} > **严重性**: {TRIGGER.SEVERITY} > **当前值**: {ITEM.VALUE} **问题详情**: {TRIGGER.DESCRIPTION} **处理建议**: {TRIGGER.COMMENT}4.2 常见问题排查
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| 收不到告警 | 脚本执行权限不足 | chmod +x /usr/lib/zabbix/alertscripts/wecom_alert.py |
| 企业微信返回错误 | access_token过期 | 检查token刷新逻辑，确保每次使用前获取最新token |
| 部分用户收不到 | 收件人配置错误 | 检查企业微信用户ID是否正确，确保用户在企业微信可见范围内 |
| 消息内容乱码 | 编码问题 | 在脚本中添加# -*- coding: utf-8 -*-，确保使用UTF-8编码 |
4.3 性能优化建议
- Token缓存：实现token本地缓存，避免每次告警都重新获取
- 消息队列：高频率告警场景下，建议引入Redis等队列系统缓冲消息
- 失败重试：在脚本中添加重试逻辑，应对网络波动情况
# 带重试机制的发送示例 def send_with_retry(url, data, max_retries=3): for attempt in range(max_retries): try: response = requests.post(url, json=data, timeout=5) return response.json() except Exception as e: if attempt == max_retries - 1: raise time.sleep(2 ** attempt)5. 实战案例：CPU负载告警配置
5.1 创建监控项
- 进入配置 → 主机 → 选择目标主机 → 监控项
- 创建新的监控项：
- 名称：CPU Load Average
- 键值：system.cpu.load[all,avg1]
- 更新间隔：1m
- 应用集：CPU
5.2 设置触发器
创建触发器表达式：
{host:system.cpu.load[all,avg1].last()}>5配置触发器参数：
- 严重性：严重
- 问题事件生成模式：多重
- 描述：
主机 {HOST.NAME} CPU负载过高，当前值为 {ITEM.VALUE} 可能原因：进程异常、资源不足 建议检查：top命令查看进程占用情况
5.3 关联告警动作
在之前创建的动作中，添加新的条件：
- 触发器= CPU Load Average
- 维护状态≠ 在维护中
测试时可通过以下命令触发告警：
# 在目标主机执行 stress-ng --cpu 4 --timeout 300s