<< 为了构建这样一个监控系统，您可以通过编写Python脚本来实现定期检查MySQL中的慢查询日志，并将结果发送到企业微信。具体步骤如下： ### 步骤一：配置 MySQL 慢查询 确保您的 `mysqld` 已经正确地开启了慢查询功能。可以在 Docker 部署时，修改或挂载自定义的 my.cnf 文件设置以下参数（如果使用的是官方镜像，请将其添加至 `/etc/mysql/conf.d/` 下）。 ```ini [mysqld] slow_query_log = 1 # 打开慢查询记录开关 long_query_time = 2 # 设置超过两秒即视为“慢” log_output=TABLE # 将慢查询存入数据库表中(便于后续读取) ``` ### 步骤二：编写 Python 监控程序 接下来是定时任务的核心逻辑代码示例，该部分会每五分钟执行一次并从 MySQL 中提取最新的慢查询信息然后推送到企业微信上。需要注意安装必要的依赖如 PyMySQL 和 wechatwork_sdk 或者其他能向企微推送消息的方法库。 #### 主要流程描述: - 使用 pymysql 连接到 docker 容器内的 MySQL 数据库； - 查询 performance_schema.slow_log 表获取最近五分钟后的新数据； - 如果存在新的慢查询，则构造请求体并通过 POST 请求的方式调用 Webhook API 发送通知给指定的企业微信群聊； ```python import time, json, requests from datetime import timedelta,datetime import pymysql # 自行替换为实际值 DB_HOST="localhost" DB_USER="root" DB_PASSWORD="" MYSQL_ROOT_PASSWORD="" CORPID='' # CorpID是企业号的标识 AGENTID='1000002' # 应用id SECRET='' # Secret可在应用详情页查看和复制 def get_slow_queries(): conn=pymysql.connect(host=DB_HOST,user=DB_USER,password=DB_PASSWORD,database='performance_schema',charset='utf8') cursor=conn.cursor() try: sql=f""" SELECT start_time, digest_text as query, timer_wait / 1000000 AS duration_ms FROM events_statements_history_long WHERE event_name LIKE 'statement/sql/%' AND timer_wait > {int(float(timedelta(seconds=2).total_seconds())*1e6)} ORDER BY start_time DESC; """ cursor.execute(sql) return [{"start":record.strftime("%Y-%m-%d %H:%M:%S"), "query": record[1], "duration": f"{record:.2f} ms"} for record in cursor.fetchall()] finally: cursor.close() conn.close() def send_msg_to_weixin(msg): token_url=f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORPID}&corpsecret={SECRET}" access_token=requests.get(token_url).json().get('access_token') if not access_token: raise Exception("Failed to fetch access token") wx_push_api=f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}' response=requests.post(wx_push_api,json.dumps({ "touser":"@all", # 接收者的userid列表 | @all表示所有成员都会收到通知 "msgtype":"text", "agentid": AGENTID, "text":{"content": msg}, "safe":"0" }),headers={"Content-Type": "application/json"}) if __name__ == "__main__": while True: slow_logs=get_slow_queries() or [] current_time=datetime.now().strftime('%Y-%m-%d %H:%M:%S') message='\n'.join([current_time]+[str(log)for log in slow_logs]) print(message) if len(slow_logs)>0:send_msg_to_weixin(message) time.sleep(300)#sleep five minutes before next check. ``` > 注意事项： > >- 修改上述脚本中的连接字符串和其他敏感字段前缀以匹配实际情况。 >- 根据需求调整 `long_query_time` 参数以及睡眠间隔 (这里设定了5分钟) 来改变轮询频率。 >- 上述脚本没有考虑安全性和异常处理，建议根据生产环境做适当优化改进。 >