python multiprocessing.Pool实例

作者：容易        2013-04-08    16:07:58

使用python进程池模块  multiprocessing.Pool 对oracle进行压力测试,简单代码实现

import time,cx_Oracle,multiprocessing,random,string 
import sys
def m_sql(x):
          db_pool = cx_Oracle.SessionPool(user="syslog",password="oracle",dsn="192.168.10.205:1521/syslog",min=2,max=5,increment=1)
#        sql_type=random_sql_type()
#        sql=random_sql(sql_type)
          sql=("insert into test values(:id,:name)",{'id':100,'name':'tiger'})
          for i in xrange(x):
                  conn = db_pool.acquire()
                  cursor = conn.cursor()
                  cursor.prepare(sql[0]) 
                  cursor.execute(None,sql[1])
                  conn.commit()
                  cursor.close()
                  db_pool.release(conn)
def random_sql(sql_type):
        sql_type=sql_type
        global cate_set
        global author_set
        if sql_type[1] == 'id':
                values={}
                values['id']=random.randint(100,100000)
                return (sql_type[0],values)
        elif sql_type[1] == 'name':
                values={}
                values['name']=str(random.choice(cate_set)[0])
                return (sql_type[0],values)
        else:
                values={}
                cate_id=random.choice(cate_set)
                values['category_id']=cate_id[0]
                values['sub_category_id']=cate_id[1]
                values['author_name']=random.choice(author_set)[0]
                values['name']=string.join(random.sample(['a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','ora','py','dj','lin','net','nginx','http','ww','ftp','cdev','tig','bin','dns','ad','exchange','sql','dev','qq','sina','haha'], 4)).replace(" ","")
                values['price']=random.randint(1,100)
                values['content']=string.join(random.sample(['lin','net','nginx','http','ww','ftp','cdev','tig','bin','dns','ad','exchange','sql','dev','qq','sina','haha','a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','ora','py','dj','lin','net','nginx','http','ww','ftp','cdev','tig','bin','dns','ad','exchange','sql','dev','qq','sina','haha'], 5)).replace(" ","")
                values['repertory']=random.randint(10000,100000)
                return (sql_type[0],values) 
def random_sql_type():
    sql_insert=("insert into blog (category_id,sub_category_id,author_name,name,price,content,repertory) values (:category_id,:sub_category_id,:author_name,:name,:price,:content,:repertory)","insert")
    return sql_insert
if __name__ == "__main__": 
        db_pool = cx_Oracle.SessionPool(user="syslog",password="oracle",dsn="192.168.10.205:1521/syslog",min=2,max=5,increment=1)  
        pro_pool = multiprocessing.Pool(processes=100)
        conn = db_pool.acquire()
        cursor = conn.cursor()
        cate_sql='select category_id,id from sub_category'
        cursor.execute(cate_sql)
        cate_set=cursor.fetchall()
        cursor.close()
        db_pool.release(conn)
        conn = db_pool.acquire()
        cursor = conn.cursor()
        sql_author='select name from author'
        cursor.execute(sql_author)
        author_set=cursor.fetchall()
        cursor.close()
        db_pool.release(conn)
        print type(db_pool)
#       pro_pool.map(m_sql,db_pool)
        for i in xrange(100):  
                pro_pool.apply_async(m_sql,args=(1000000,)) 
        pro_pool.close()
        pro_pool.join()
        print "insert into 0K!,sleep 60 seconds!"
        print "Sub-process(es) done."