最近公司做數據庫的單台服務器多實例，需要對各個數據庫的進程進行監控，因為我們的資產數據庫中已經有這個服務器跑多少個實例，端口是什麼，用來做什麼的，所以就拿資產數據庫的端口和服務器本地運行的數據庫端口進行對比，報出沒有運行的mysql實例以及他的用途。一種方法是根據"ps auxww|grep mysqld|grep -v root|grep -v grep"這個命令抓取本地運行的數據庫端口，不過你也可以根據netstat命令來獲取本地數據庫實例的所有端口；還有一種方法就是從資產中得到這個服務器應該運行的所有mysql端口，用python的socket模塊來檢測端口是否存活，這種方法比較簡單一些。我用第一種方法是因為我的這段代碼我已經寫過用於其他用途，等於復用，圖個省事。以下是代碼內容：
#!/bin/env python
# -*- coding: utf-8 -*-
import os,sys,MySQLdb
def center(sql):#連接數據庫
try:
center_ip = '192.168.1.100'
center_user = 'root'
center_passwd = 'xxxxxx'
conn = MySQLdb.connect(host = center_ip,user = center_user,passwd = center_passwd,charset='utf8',connect_timeout=20)
cursor = conn.cursor()
cursor.execute(sql)
alldata = cursor.fetchall()
cursor.close()
conn.close()
return alldata
except:
return 0
class check_port():#走資產中獲取本地IP中應該有多少個mysql實例端口
def __init__(self):
conn = "ip a|grep glob|grep -v '192.168'|awk '{print $2}'"
self.host = os.popen(conn).readlines()[0].split("/")[0]
def remot(self):
sql = "SELECT PORT FROM center.host_info WHERE flag='1' AND del_info!=0 AND ip='%s' AND host_name IN ('a','b','c')" % self.host
alldata = center(sql)
cent_port = []
if alldata != 0:
for i in alldata:
cent_port.append(str(i[0]))
return cent_port
else:
return cent_port
def local(self):#獲取本地mysql有多少個實例運行
psinfo = os.popen("ps auxww|grep mysqld|grep -v root|grep -v grep").readlines()
local_port = []
if not psinfo:
return local_port
for i in psinfo:
for j in i.split("--"):
if j.find("port") != -1:
port = j.split("=")[1].strip()
local_port.append(port)
return local_port
def main(self):
local_port = self.local()
cent_port = self.remot()
cent_port.sort()
local_port.sort()
if local_port == cent_port and len(local_port) != 0 and len(cent_port) != 0:
print 0
else:
error = ""
diff_list = list(set(local_port) ^ set(cent_port))
for port in diff_list:
sql = "SELECT CONCAT(a.main_name,'_',b.sub_name,'_',c.app_name,'_',c.port) AS used FROM center_app.main_category a, center_app.sub_category b, center_app.app_info c WHERE a.id = b.main_id AND b.dist_id = c.dist_id AND b.main_id = c.main_id AND b.main_id='2' AND c.flag='1' AND c.del_info!=0 AND c.ip='%s' and c.port='%s'" % (self.host,port)
alldata = center(sql)
if error == "":
error = error + alldata[0][0]
else:
error = error + ";" + alldata[0][0]
print error
if __name__ == "__main__":
boss = check_port()
boss.main()
如果用第二種方法的話，很簡單，用下面的函數可以實現這個端口測試：
ipmort socket
def test_port()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
address = '127.0.0.1'
port = 80
try:
s.connect((address,port))
return True
except Exception,e:
return False
s.close()