最近公司做数据库的单台服务器多实例，需要对各个数据库的进程进行监控，因为我们的资产数据库中已经有这个服务器跑多少个实例，端口是什么，用来做什么的，所以就拿资产数据库的端口和服务器本地运行的数据库端口进行对比，报出没有运行的mysql实例以及他的用途。一种方法是根据"ps auxww|grep mysqld|grep -v root|grep -v grep"这个命令抓取本地运行的数据库端口，不过你也可以根据netstat命令来获取本地数据库实例的所有端口；还有一种方法就是从资产中得到这个服务器应该运行的所有mysql端口，用python的socket模块来检测端口是否存活，这种方法比较简单一些。我用第一种方法是因为我的这段代码我已经写过用于其他用途，等于复用，图个省事。以下是代码内容：

1. #!/bin/env python 
2. # -*- coding: utf-8 -*- 
3.  
4. import os,sys,MySQLdb 
5.  
6. def center(sql):#连接数据库 
7.     try: 
8.         center_ip = '192.168.1.100' 
9.         center_user = 'root' 
10.         center_passwd = 'xxxxxx' 
11.         conn = MySQLdb.connect(host = center_ip,user = center_user,passwd = center_passwd,charset='utf8',connect_timeout=20) 
12.         cursor = conn.cursor()  
13.         cursor.execute(sql) 
14.         alldata = cursor.fetchall() 
15.         cursor.close() 
16.         conn.close() 
17.         return alldata 
18.     except: 
19.         return 0 
20.  
21. class check_port():#走资产中获取本地IP中应该有多少个mysql实例端口 
22.  
23.     def __init__(self): 
24.         conn = "ip a|grep glob|grep -v '192.168'|awk '{print $2}'" 
25.         self.host = os.popen(conn).readlines()[0].split("/")[0] 
26.  
27.     def remot(self): 
28.         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 
29.         alldata = center(sql) 
30.         cent_port = [] 
31.         if alldata != 0: 
32.             for i in alldata: 
33.                 cent_port.append(str(i[0])) 
34.             return cent_port 
35.         else: 
36.             return cent_port 
37.  
38.     def local(self):#获取本地mysql有多少个实例运行 
39.         psinfo = os.popen("ps auxww|grep mysqld|grep -v root|grep -v grep").readlines() 
40.         local_port = [] 
41.         if not psinfo: 
42.             return local_port 
43.         for i in psinfo: 
44.             for j in i.split("--"): 
45.                 if j.find("port") != -1: 
46.                     port = j.split("=")[1].strip() 
47.                     local_port.append(port) 
48.         return local_port 
49.  
50.     def main(self): 
51.         local_port = self.local() 
52.         cent_port = self.remot() 
53.                 cent_port.sort() 
54.                 local_port.sort() 
55.         if local_port == cent_port and len(local_port) != 0 and len(cent_port) != 0: 
56.             print 0 
57.         else: 
58.             error = "" 
59.             diff_list = list(set(local_port) ^ set(cent_port)) 
60.             for port in diff_list: 
61.                 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) 
62.                 alldata = center(sql) 
63.                 if error == "": 
64.                     error = error + alldata[0][0] 
65.                 else: 
66.                     error = error + ";" + alldata[0][0] 
67.             print error 
68.  
69. if __name__ == "__main__": 
70.     boss = check_port() 
71.     boss.main() 

 如果用第二种方法的话，很简单，用下面的函数可以实现这个端口测试：

1. ipmort socket 
2.  
3. def test_port() 
4.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5. s.settimeout(2) 
6.     address = '127.0.0.1' 
7.     port = 80 
8.     try: 
9.         s.connect((address,port)) 
10.         return True 
11.     except Exception,e: 
12.         return False     
13. s.close()