在Python中如何设计多线程？

最近做了一个sqlServer下的大批量插入的多线程处理。方法如下。

handArrays.py

import json

import os

from os import path

import sys

import time

import threading

from datetime import datetime

farther_path = os.path.abspath(os.path.join('..')) #父路径

current_path=sys.path[0]

urls_path=os.path.abspath(os.path.join('..\\url'))

sys.path.append(urls_path)

from getPaths import GetPath

db_path=os.path.abspath(os.path.join('..\\db'))

sys.path.append(db_path)

from Conn import GetConn

from getInfos import GetInfo

class HandArray():

def __init__(self,arrays,tables,database,databasetype):

self.arrays=arrays

self.tables=tables #表英文名

self.database=database

self.databasetype=databasetype

self.cols=len(self.arrays[0]) #提供数组列数目

listInfo=GetInfo(self.database,self.databasetype,self.tables)

strcols=listInfo.infos[0][0]

self.lens=listInfo.length #表列数目

self.tablesinfo='insert into '+tables+' ('+strcols+') values'

self.db=GetConn(self.database,self.databasetype).db

self.cursor =self.db.cursor()

def handleArrays(self,x,y ):

# tablesinfo 插入目标insert 头信息

array=self.arrays[x-1:y]

i=0

j=0

rows=y+1-x #行数

zkh=''

ykh=''

dh=''

rowl=''

val=''

for i in range(0,rows):

for j in range(0,self.cols):

if j == 0 :

zkh='('

dh=''

else :

zkh=''

dh=''

if j == self.cols-1 and i!=rows-1:#末尾列非最后一行

ykh='),'

dh=''

elif j == self.cols-1 and i==rows-1: #末尾列最后一行

ykh=');'

dh=''

else :

ykh=''

dh=','

val=val+zkh+"'"+str(array[i][j])+"'"+dh+ykh

sqls=self.tablesinfo+val

try:

self.cursor.execute(sqls)

self.db.commit()

except:

print(sqls[-2:2],'异常结束啦')

def closedb(self):

try:

self.cursor.close()

self.db.close()

print('end关闭数据结束---------------------')

except:

print('事务出错')

arraythreads.py

import threading

from concurrent.futures import ThreadPoolExecutor

import json

import os

from os import path

import sys

from datetime import datetime

farther_path = os.path.abspath(os.path.join('..')) #父路径

current_path=sys.path[0]

urls_path=os.path.abspath(os.path.join('..\\url'))

sys.path.append(urls_path)

from getPaths import GetPath

db_path=os.path.abspath(os.path.join('..\\db'))

sys.path.append(db_path)

from Conn import GetConn

from getInfos import GetInfo

from handArrays import HandArray

class ArrayThread():

def __init__(self,arrays):

self.lends=0

self.arrays=arrays

self.indexMaps=self.getIndexMap()

def getIndexMap(self):

length=len(self.arrays)

ds={}

j=0

i=1

lends =0

while(j< length ):

s=(j+1)%8

if s==0:

ds[i] = j

lends = i

i=i+1

j=j+1

if len(ds)>0:#

if ds[lends]<length:

ds[lends+1]=length

lends=lends+1

elif len(ds)==0 and length>0:

ds[1] = length

lends=1

self.lends=lends

print(ds)

return ds

def startThread(self,ha):#并根据索引键值对多线程处理 arrays,cols,tablesinfo

thread = [threading.Thread() for col in range(self.lends)]

length=self.lends #定18个元素

bg=1

ed=0

idx=1

for i in range(1, length+1):

ed=self.indexMaps[i]

thread[i-1]=threading.Thread( name ='t'+ str(i-1), target=ha.handleArrays,args=(bg,ed))

thread[i-1].setDaemon(True)

thread[i-1].start()

thread[i-1].join()

idx=ed +1

bg=idx

def action(max):

my_sum = 0

for i in range(max):

print(threading.current_thread().name + ' ' + str(i))

my_sum += i

return my_sum

if __name__ == '__main__':

def createArrays( types ,rows , cols ):

arrays =[[ types(0) for col in range( cols )] for row in range(rows )]

j=1

for m in range(0,rows):

for n in range(0,cols):

arrays[m][n]=str(j)

j=j+1

print('建数组',arrays[rows-1][cols-1],'建数组结束')

return arrays

arrays = createArrays(str,100,5) #插入100条数据

ha=HandArray(arrays,'t_test','stocks','MSSQL')#创建连接 提供基础信息

th=ArrayThread(arrays)

th.startThread(ha)

ha.closedb()

该算法实现了多线程分割数组，将大量数据高效插入sqlserver数据库。需要说明的是，由于该算法无限制创建了线程，可能会导致插入的时候数据丢失，因此在新的算法中，我们可以加入线程池，用队列管理有限个线程来实现最终效果。

线程池设计如下：