[python] 连接MySQL，以及多线程、多进程连接MySQL初探

rger

                      环境：Linux CentOS6.7，python 2.7.13
说明：连接MySQL，进行增删改查操作，并将执行的SQL和耗时记录到日志里
demo:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123

#!/usr/bin/env python # -*- coding:utf-8 -*- import MySQLdb import logging import time ''' 设置日志输出路径和格式 ''' logging.basicConfig(level=logging.INFO,                     format='%(asctime)s %(levelname)s %(message)s',                     datefmt='%Y-%m-%d %H:%M:%S',                     filename='/home/sql.log',                     filemode='a+') ''' 执行的SQL语句可能有多个，可以在这边进行组装 ''' def sql_assemble(*args):     sql_list = [         'delete from table_a where id = %s' % args[0],         'delete from table_b where content in %s' % args[1],         'update table_c set aa = %s' % args[2],     ]     return sql_list class mysqlopr(object):     def __init__(self,ip,port,username,password,dbname,char_set='utf8'):     '''     创建对象的时候就要求将参数传入进来     '''         self.ip = ip         self.port = port         self.username = username         self.password = password         self.dbname = dbname         self.char_set = char_set     def connect(self):     '''     mysql如果开启了安全模式，insert、delete和update无法执行，需要先设置SET SQL_SAFE_UPDATES = 0     执行单条SQL，使用execute(),执行多条SQL，可以使用executemany()，但只有INSERT和REPLACE会执行的快点。说明如下     executemany：This method improves performance on multiple-row INSERT and REPLACE. Otherwise it is equivalent to looping over args with execute().     '''         self.db = MySQLdb.connect(host=self.ip,port = self.port,user=self.username,passwd=self.password,db =self.dbname,charset=self.char_set)         self.cursor = self.db.cursor()         #sql_s = 'SET SQL_SAFE_UPDATES = 0'         #self.cursor.execute(sql_s)         logging.info('link to mysqldb: '+self.dbname)     def select(self,sql):     '''     sst :查询开始时间     set :查询结束时间     fetchone(): 只返回一行     fetchall(): 返回所有行     fetchmany(size): 返回size行     rowcount: 这是一个long型只读属性，返回执行execute()方法后影响的行数。               '''         try:             logging.info(sql)             self.index_list = []             sst = time.time()             self.cursor.execute(sql)             set = time.time()             #select_result = self.cursor.fetchone()             #select_result = self.cursor.fetchmany(10)             select_result = self.cursor.fetchall()             #print self.cursor.rowcount,type(self.cursor.rowcount)             logging.info('select count: ' + self.cursor.rowcount + ', cost :' + str(sst - set))         except Exception as e:             print e             logging.error("Error: unable to fecth data" + str(e))     def insert(self):     '''     和其他方法类似，这边省略     '''         pass     def update(self):     '''     和其他方法类似，这边省略     '''         pass     def delete(self,sql):     '''     dst:删除操作开始时间     det:删除操作结束时间     删除成功提交，失败回滚     '''         try:             logging.info(sql)             dst = time.time()             self.cursor.execute(sql)             self.commit()             det = time.time()             logging.info('delete row: ' + str(self.cursor.rowcount) + ', cost :' + str(det - dst))         except Exception as e:             print e             self.rollback()             logging.error(str(sql) + ',' + str(e))     def commit(self):         self.db.commit()     def rollback(self):         self.db.rollback()     def close(self):        self.cursor.close()        self.db.close() if __name__ == '__main__':     st = time.time()     a = mysqlopr(ip='192.168.1.1',port = 8888,username='try',password='123456',dbname ='trydb',char_set='utf8')     a.connect()     '''     如果参数相同，SQL语句不同，可以现在sql_assemble方法中配置好SQL，再把参数传递进去。     或者这边直接调用对应的方法执行     这里过程省略     '''     a.close()     et = time.time()     logging.info("SQL executed over,cost: " + str(et -st))     print et - st

多线程、多进程连接MySQL说明：
1、多线程连接：
起初我使用threading.thread模块，先建立一个MySQL连接，然后由多个线程来执行具体的SQL。但发现在执行的时候，不是报MySQL连接被关闭，就是出现其他异常错误。上网查询，是因为多个线程无法共享一个数据库连接，会出现不可预测的情况，官方文档说明如下：
The MySQL protocol can not handle multiple threads using the same connection at once. Some earlier versions of MySQLdb utilized locking to achieve a threadsafety of 2. While this is not terribly hard to accomplish using the standard Cursor class (which uses mysql_store_result()), it is complicated by SSCursor (which uses mysql_use_result(); with the latter you must ensure all the rows have been read before another query can be executed. It is further complicated by the addition of transactions, since transactions start when a cursor execute a query, but end when COMMIT or ROLLBACK is executed by the Connection object. Two threads simply cannot share a connection while a transaction is in progress, in addition to not being able to share it during query execution. This excessively complicated the code to the point where it just isn't worth it.


The general upshot of this is: Don't share connections between threads. It's really not worth your effort or mine, and in the end, will probably hurt performance, since the MySQL server runs a separate thread for each connection. You can certainly do things like cache connections in a pool, and give those connections to one thread at a time. If you let two threads use a connection simultaneously, the MySQL client library will probably upchuck and die. You have been warned.

For threaded applications, try using a connection pool. This can be done using the Pool module.

官方建议使用连接池模块，参照了他人的做法，使用DBUtils.PooledDB来创建线程连接池，一次性创建多个连接，但是查询MySQL中的连接执行SQL语句情况，发现虽然建立了多个连接，但是只有一个连接有在执行SQL，其他连接都是sleep状态，至今不明白为何是这情况。网上的例子都是抄来抄去的，没人对使用DBUtils模块连接MySQL执行多线程操作的情况及效率做进一步的研究和分析。

2、多进程连接：
后面没辙，改用多进程（multiprocessing模块），可以和MySQL建立多个连接并发执行SQL。对比执行耗时，整体性能比单个进程快，但其中单个SQL的执行效率，多进程没有没有单进程执行的快。
demo:

1 2 3 4 5 6 7 8 9 10 11 12 13 14

#!/usr/bin/env python # -*- coding:utf-8 -*- import MySQLdb from multiprocessing import Process,Pool class mysqlopr():     '''省略''' pool = Pool(5)        ####设置进程数 for i in range(10):     pool.apply_async(func=run_sql_func, args=(arg,))       ####异步执行     #pool.apply(func=run_sql_func, args=(arg,))                    ####同步执行，官方不建议使用，python3.+版本已无该方法 pool.close() pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。