步骤:启动一个单线程,循环执行连接数据库的任务,每隔2秒执行一次连接,如果连接不上MYSQL,产生错误,这时候可以通知备机启动服务接管主机任务或者利用winsound模块让你的电脑发出报警音。
import threadingimport timeimport pymysql
def mysqltask(): while True: try: # 尝试连接到MySQL服务 conn = pymysql.connect(host='localhost', user='root', password='123456', database='test') # 如果连接成功,服务正在运行 print("MySQL service is running."+time.ctime()) #关闭连接 conn.close() except pymysql.err.OperationalError: # 如果连接失败,服务可能没有运行 print("MySQL service is not running."+time.ctime()) break #每两秒执行一次连接 time.sleep(2)
if __name__ == '__main__': checkmysql = threading.Thread(target=mysqltask) checkmysql.start()
threading是线程模块
time时间模块
pymysql操作mysql模块(pyodbc操作sqlserver模块)
效果:
领取专属 10元无门槛券
私享最新 技术干货