实践 2-2 python多线程读写mysql数据库

　　关键字：python多线程、mysql数据库连接、python数据库连接池

1、python多线程

理论部分大部分参考他人，不过想写简单点，也就不够全面，更详细的可查阅参考链接的前三个。

1.1 python多线程

为什么要使用多线程？是为了提高程序运行效率，也希望一些耗时过长的操作(如数据库访问、网络访问、文件操作等)不会阻塞主线程。

但是由于GIL锁，python的多线程并不能算作真正的多线程，GIL（Global Interpreter Lovk，全局解释器）限制在任意时刻只能由一个线程在解释器中运行。所以python多线程并不适合CPU密集型的程序(反而会因为线程切换耗费更多的时间)，在I/O密集型程序中比较适用。

python线程不算作并行运行在多核cpu中，只能算作并发运行，并不能充分利用cpu资源。若是要实现真正的并行，则需要考虑multiprocessing--Python多进程编程。

1.2 threading模块

在Python中，我们一般使用threading模块来实现多线程操作。

threading模块中包含了关于线程操作的丰富功能，包括：常用的线程函数、线程对象(Thread)、锁对象(Lock)、递归锁对象(RLock)、事件对象(Event)、条件变量对象(Condition)、信号量对象(Semaphore)、定时器对象(Timer)、栅栏对象(Barrier)等。

threading中常用的线程函数包含：

threading.currentThread(): 返回当前的线程变量。
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

其他相关的对象，这里仅涉及Thread、Lock和RLock，以及同步、线程安全的队列Queue。

1.3 threading.Thread

可以使用threading.Thread来创建线程，有两种方式：

直接创建
继承创建

1.3.1 直接创建

可以直接向threading.Thread传入可调用函数。

#!/usr/bin/pyhon
#coding=utf-8


import time
import random
import threading


def func(name):
    s = random.randint(1, 5)
    print(f'current thread is {name}, sleeping {s}s.')
    time.sleep(s)
    print(f'thread {name} is over')


if __name__ == '__main__':
    for i in range(1, 5):
        t = threading.Thread(target=func, args=(i,))
        t.start()
    print('Main Thread')

1.3.2 继承创建

继承threading.Thread，重写run方法。

import time
import random
import threading


class Func(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        s = random.randint(1, 5)
        print(f'current thread is {self.name}, sleeping {s}s.')
        time.sleep(s)
        print(f'thread {self.name} is over')


if __name__ == '__main__':
    for i in range(1, 5):
        t = Func(str(i))
        t.start()
    print('Main Thread')

1.3.3 Thread构造参数

threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

下面是Thread的参数说明

group：默认为None（该参数是为了以后实现ThreadGroup类而保留的）
target：在run方法中调用的可调用对象，即需要开启线程的可调用对象，比如函数或方法。
name：线程名称，默认为“Thread-N”形式的名称，N为较小的十进制数。
args：在参数target中传入的可调用对象的参数元组，默认为空元组()。
kwargs：在参数target中传入的可调用对象的关键字参数字典，默认为空字典{}。
daemon：默认为None，即继承当前调用者线程（即开启线程的线程，一般就是主线程）的守护模式属性，如果不为None，则无论该线程是否为守护模式，都会被设置为“守护模式”。

1.3.4 Thread常用方法

start()：开启线程活动。它将使得run()方法在一个独立的控制线程中被调用，需要注意的是同一个线程对象的start()方法只能被调用一次，如果调用多次，则会报RuntimeError错误。
run()：此方法代表线程活动。
join(timeout=None)：让当前调用者线程（一般为主线程）等待，直到线程结束。
daemon：表示该线程是否为守护线程，True或者False。设置一个线程的daemon必须在线程的start()方法之前，否则会报RuntimeError错误。这个值默认继承自创建它的线程，主线程默认是非守护线程，所以在主线程中创建的线程默认都是非守护线程的，即daemon=False。

1.3.5 守护线程

有一种线程，它是在后台运行的，它的任务就是为其他线程提供服务，这种线程被称为“”后台线程（Daemon Thread），又称为“守护线程”或“精灵线程”。python解释器的垃圾回收线程就是典型的后台线程。

后台线程有一个特征，如果所有的前台线程都死亡了，那么后台线程会自动死亡。

主线程默认是前台线程，由前台线程创建的未设置daemon为True的线程也是前台线程（默认）。设置了daemon的线程是后台线程，由后台线程创建的子线程也默认是后台线程。

1.4 Lock和RLock

threading 模块提供了 Lock 和 RLock 两个类，它们都提供了如下两个方法来加锁和释放锁：

acquire(blocking=True, timeout=-1)：请求对 Lock 或 RLock 加锁，其中 timeout 参数指定加锁多少秒。
release()：释放锁。

Lock 和 RLock 的区别如下：

threading.Lock：它是一个基本的锁对象，每次只能锁定一次，其余的锁请求，需等待锁释放后才能获取。
threading.RLock：它代表可重入锁（Reentrant Lock）。对于可重入锁，在同一个线程中可以对它进行多次锁定，也可以多次释放。如果使用 RLock，那么 acquire() 和 release() 方法必须成对出现。如果调用了 n 次 acquire() 加锁，则必须调用 n 次 release() 才能释放锁。

Lock 和RLock 的使用方法相似，但是常用的是RLock(可重入锁)，因为相比于Lock，Rlock在同一线程中函数嵌套调用同一个锁锁定区域的情况下，可以防止死锁，具体例子如下。

import time
import random
import threading

class My_Thread(threading.Thread):
    def __init__(self,lock):
        super().__init__()
        self.setDaemon(True)
        self.lock = lock
        
    def run(self):
        self.lock.acquire()
        print(self.name,'获取锁，并进入共享区域f()')
        self.g()
        self.lock.release()
        print(self.name,'退出共享区域f()，并释放锁')

    def g(self):
        self.lock.acquire()
        print(self.name,'获取锁，并进入共享区域g()')
        
        self.lock.release()
        print(self.name,'退出共享区域g()，并释放锁')

if __name__ == '__main__':
    #lock = threading.Lock()
    lock = threading.RLock()
    t = My_Thread(lock)
    t.start()
    print('主线程结束')

结果如下所示，在一个线程中嵌套请求锁资源，当使用Lock()时线程形成死锁，并未返回正常的结果。而使用Rlock()会允许同一个线程进入已经持有锁的区域不需要请求锁、等待锁，只是维持一个计数。在一个线程中维持好请求锁和释放锁的对应关系，在该线程完成所有锁定资源的操作后计数为0就可以释放锁，供其他线程使用。

##lock = threading.Lock()
Thread-1 获取锁，并进入共享区域f()
主线程结束

##lock = threading.RLock()
Thread-1 获取锁，并进入共享区域f()
Thread-1 获取锁，并进入共享区域g()
Thread-1 退出共享区域g()，并释放锁
Thread-1 退出共享区域f()，并释放锁
主线程结束

为什么要使用守护线程？

~~如果代码中形成死锁会不会对电脑造成永久影响，在线等挺急的！~~

~~也不急，反正应该进程结束的时候所有线程都会结束，再不济关个机也差不多了。~~

所以还是安全为上。

1.5 线程优先级队列Queue

python的Queue模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出）队列Queue、LIFO（后入先出）队列LifoQueue，和优先级队列PriorityQueue。

这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程间的同步。

Queue 模块中的常用方法:

Queue.qsize() 返回队列的大小
Queue.empty() 如果队列为空，返回True,反之False
Queue.full() 如果队列满了，返回True,反之False
Queue.full 与 maxsize 大小对应
Queue.get([block[, timeout]])获取队列，timeout等待时间
Queue.get_nowait() 相当Queue.get(False)
Queue.put(item) 写入队列，timeout等待时间
Queue.put_nowait(item) 相当Queue.put(item, False)
Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
Queue.join() 实际上意味着等到队列为空，再执行别的操作

#!/usr/bin/python3

import queue
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        print ("开启线程：" + self.name)
        process_data(self.name, self.q)
        print ("退出线程：" + self.name)

def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print ("%s processing %s" % (threadName, data))
        else:
            queueLock.release()
        time.sleep(1)

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = queue.Queue(10)
threads = []
threadID = 1

# 创建新线程
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1

# 填充队列
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()

# 等待队列清空
while not workQueue.empty():
    pass

# 通知线程是时候退出
exitFlag = 1

# 等待所有线程完成
for t in threads:
    t.join()
print ("退出主线程")

1.6 python线程池

1.6.1 线程池

系统启动一个新县城的成本是很高的，因为它涉及与操作系统的交互。在这种情况下，使用线程池可以很好地提升性能，尤其是当程序中小创建大量生存期很短暂的线程时，更应该考虑使用线程池。

线程池在系统启动时即创建大量空闲的线程，程序只要将一个函数提交给线程池，线程池就会启动一个空闲的线程来执行它。当该函数执行结束后，该线程并不会死亡，而是再次返回线程池中编程空闲状态，等待下一个函数。

此外，使用线程池可以有效地控制系统中并发线程的数量，当系统中包含有大量的并发线程时，会导致系统性能急剧下降，甚至导致python解释器崩溃，而线程池的最大线程数参数可以控制系统中并发线程的数量不超过此数。

1.6.1 线程池的使用

在过去，我们可以使用第三方模块threadpool来创建线程池，但是现在主流的使用线程池的模块是python3中自带模块concurrent.futures模块中的ThreadPoolExecutor。

以下是ThreadPoolExcutor的使用方法。

import time
import random
from concurrent.futures import ThreadPoolExecutor


def func(name):
    s = random.randint(1, 5)
    print(f'current thread is {name}, sleeping {s}s.')
    time.sleep(s)
    print(f'thread {name} is over')


if __name__ == '__main__':
    with ThreadPoolExecutor(max_workers=3) as t:
        for i in range(1, 6):
            t.submit(func, i)

2、python连接mysql数据库

java连接mysql数据库在：实践 1-1 JDBC使用详解。

python连接Oracle数据库在：实践2-1 python连接Oracle数据库

2.1 环境配置-PyMySQL

PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库，Python2中则使用mysqldb。如果习惯使用mysqldb，可以使用下面的语句导入使用。这里仅讨论PyMySQL 。

PyMySQL参考文档：欢迎使用 PyMySQL 的文档！

import pymysql
pymysql.install_as_MySQLdb()

PyMySQL安装

pip3 install pymysql

疑惑：一开始我用的是pip3 install PyMySQL,在pip list里的名称也是PyMySQL（安装时使用两种没差别），但是如果使用import PyMySQL则会报没有模块，只能使用import pymysql，这是为什么？对python的模块引入规则还比较陌生。

myslq数据库配置，建表如下：

CREATE TABLE `b_table` (
  `id` int(11) NOT NULL,
  `item` varchar(255) DEFAULT NULL,
  `time` varchar(255) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8

已有的数据量如下，id是递增序列，item是随意字符，time是当时时间串+id

select count(*) from b_table;--1000000

2.2 数据库操作

2.2.1 数据库连接对象

通过print(help(pymysql.connect))命令或者print(help(pymysql.connections.Connection))，可以看到pymysql的数据库连接pymysql.connections.Connection对象，其通过调用pymysql.connect()来获取，可传入参数如下所示。

class Connection(builtins.object)
 |  Connection(*, user=None, password='', host=None, database=None, unix_socket=None, port=0, charset='', sql_mode=None, read_default_file=None, conv=None, use_unicode=True, client_flag=0, cursorclass=<class 'pymysql.cursors.Cursor'>, init_command=None, connect_timeout=10, read_default_group=None, autocommit=False, local_infile=False, max_allowed_packet=16777216, defer_connect=False, auth_plugin_map=None, read_timeout=None, write_timeout=None, bind_address=None, binary_prefix=False, program_name=None, server_public_key=None, ssl=None, ssl_ca=None, ssl_cert=None, ssl_disabled=None, ssl_key=None, ssl_verify_cert=None, ssl_verify_identity=None, compress=None, named_pipe=None, passwd=None, db=None)

参数很多，但是主要关注的参数包含：

host：数据库服务器所在的主机
user：登录的用户名
password：登录用户的密码
database：要使用的数据库，None不使用特定数据库
port：访问的端口，默认值是3306
charset：字符集

#!/usr/bin/python3
 
import pymysql
 
# 打开数据库连接
db = pymysql.connect("localhost","testuser","test123","TESTDB" )
# do somethings ....
 
# 关闭数据库连接
db.close()

常用的方法包含：

close()：关闭数据库
commit()：提交事务
rollback()：回滚事务。
cursor(cursor=None)：创建光标对象，光标类型包含Cursor（默认）、SSCursor、DictCursor、SSDictCursor。
ping(reconnect=True)：测试连接服务是否还活动，reconnect为True的话，当连接关闭则尝试重连接，否则当连接关闭时就抛出异常。
select_db(db)：设置数据库
open：返回连接是否打开
show_warnings()：发送"SHOW WARNINGS"的SQL命令。
autocommit(value)：设置是否自动提交，默认为False。

2.2.2 光标对象

光标对象有四种

#1、 最常用的光标，是用于与数据库交互的对象
class pymysql.cursors.Cursor(connection)
#2、Unbuffered Cursor，主要用于返回大量数据的查询，或用于通过慢速网络连接到远程服务器，无法向后滚动
class pymysql.cursors.SSCursor(connection)
# 3、将结果作为字典返回的游标
class pymysql.cursors.DictCursor(connection)
# 4、Unbuffered Cursor,将结果作为字典返回的游标
class pymysql.cursors.SSDictCursor(connection)

光标一般不直接创建，而是调用connections.Connection.cursor()创建。

def test_connect(user_name='root',password='123456',db='learn',url='localhost'):
    # 创建连接
    conn = pymysql.connect(host=url,user=user_name,password=password,database=db)
    # 创建字典光标
    cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
    # 查询数据
    cur.execute('select b_table from hero limit 10')
    # 获取全部查询结果
    rs = cur.fetchall()
    # 输出结果
    print(rs)
    # 关闭光标
    cur.close()
    # 关闭连接
    conn.close()

下面主要讲一下pymysql.cursors.Cursor的方法：

colse()：关闭光标只会耗尽所有剩余数据。
execute(query,args=None)：运行语句，可传入绑定变量。
executemany(query,args)：针对一个语句格式运行多个绑定数据，args是tuple或者list格式的一连串数据。此方法提高了多行 INSERT 和 REPLACE 的性能。
fetchall()：返回查询的全部数据
fetchmany(size=None)：返回指定数量的数据
fetchone()：返回下一行数据
max_stmt_length=1024000：executemany()生成的最大语句大小。
mogridy(query,args=None)：返回将通过调用 execute() 方法发送到数据库的确切字符串。可用于检测输入是否正确。

2.2.3 查询数据

查询数据只能使用execute，获取查询结果的方法有三种fetchall()、fetchmany(size=None)和fetchone()。

def test_connect(user_name='root',password='123456',db='learn',url='localhost'):
    # 创建连接
    conn = pymysql.connect(host=url,user=user_name,password=password,database=db)
    # 创建光标
    cur = conn.cursor()
    
    # 进行查询：fetchall应用
    print('返回全部：fetchall')
    cur.execute('select * from b_table limit 10')
    rs = cur.fetchall()
    print(rs)

    # 进行查询： fetchone 应用
    print('逐个返回：fetchone')  
    cur.execute('select * from b_table limit 10')
    while(1):
        rs = cur.fetchone()
        if rs == None: 
            break
        print(rs)

    # 进行查询： fetchmany 应用    
    print('返回部分：fetchmany')
    cur.execute('select * from b_table limit 10')
    while(1):
        rs = cur.fetchmany(3)
        if len(rs) == 0: 
            break
        print(rs)
    
    # 关闭光标
    cur.close()
    # 关闭连接
    conn.close()

运行结果：

返回全部：fetchall
((0, 'aaa', '202108311525550'), (1, 'aaa', '202108311525551'), (2, 'aaa', '202108311525552'), (3, 'aaa', '202108311525553'), (4, 'aaa', '202108311525554'), (5, 'aaa', '202108311525555'), (6, 'aaa', '202108311525556'), (7, 'aaa', '202108311525557'), (8, 'aaa', '202108311525558'), (9, 'aaa', '202108311525559'))
逐个返回：fetchone
(0, 'aaa', '202108311525550')
(1, 'aaa', '202108311525551')
(2, 'aaa', '202108311525552')
(3, 'aaa', '202108311525553')
(4, 'aaa', '202108311525554')
(5, 'aaa', '202108311525555')
(6, 'aaa', '202108311525556')
(7, 'aaa', '202108311525557')
(8, 'aaa', '202108311525558')
(9, 'aaa', '202108311525559')
返回部分：fetchmany
((0, 'aaa', '202108311525550'), (1, 'aaa', '202108311525551'), (2, 'aaa', '202108311525552'))
((3, 'aaa', '202108311525553'), (4, 'aaa', '202108311525554'), (5, 'aaa', '202108311525555'))
((6, 'aaa', '202108311525556'), (7, 'aaa', '202108311525557'), (8, 'aaa', '202108311525558'))
((9, 'aaa', '202108311525559'),)

2.2.4 数据插入

数据的增删改涉及事务的提交和回滚，pymysql的自动提交可以通过conn.autocommit(True)来进行设置，默认是False。当自动提交为False的时候需要在合适的位置手动调用conn.commit()和conn.rollback()。

pymysql的数据插入可通过Cursor.execute(sql)，和Cursor.executemany(sql)来运行，Cursor.execute(sql)可直接运行拼接好的sql语句，但为了防止SQL注入，建议使用绑定变量。

样例代码如下，设置了自动提交为True，使用了绑定变量。

def test_insert(user_name='root',password='123456',db='learn',url='localhost'):
    # 创建连接
    conn = pymysql.connect(host=url,user=user_name,password=password,database=db)
    # 设置自动提交,默认是Flase
    conn.autocommit(True)
    print(conn.get_autocommit())
    # 创建光标
    cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
    
    # 查询目标表最大id
    cur.execute('select max(id) max_id from b_table')
    id = cur.fetchone()['max_id']
    id_t = id
    
    # 1--单个数据插入 execute
    cur_insert = conn.cursor()
    id_t = id_t+1
    time1 = datetime.datetime.now().strftime('%Y%m%d%H%M%H')+str(id_t)
    # 绑定变量对象可以是元组，也可以是列表
    #line = (id_t,'222',time1)
    line = [id_t,'222',time1]

    # 查看execute要运行的语句
    # print(cur_insert.mogrify('insert into b_table(id,item,time) values(%s,%s,%s)',line))
    
    # execute
    cur_insert.execute('insert into b_table(id,item,time) values(%s,%s,%s)',line)
    
    # 2---多个插入语句运行 executemany
    lines = []
    for i in range(10):
        id_t = id_t+1
        time1 = datetime.datetime.now().strftime('%Y%m%d%H%M%H')+str(id_t)
        line = (id_t,'222',time1)
        lines.append(line)
        
    # 查看execute要运行的语句
    #print(cur_insert.mogrify('insert into b_table(id,item,time) values(%s,%s,%s)',lines[0]))
    
    # executemany
    cur_insert.executemany('insert into b_table(id,item,time) values(%s,%s,%s)',lines)
    
    # 查看结果
    cur.execute('select * from b_table where id between %s and %s',[id+1,id_t])
    rs = cur.fetchall()
    print("插入结果:")
    for r in rs:
        print(r)
    
    # 关闭光标
    cur_insert.close()
    cur.close()
    # 关闭连接
    conn.close()

2.2.5 数据更新

数据更新也可以用execute或者executemany。样例代码设置了自动提交的值为False，配置了抛出异常处理。

def test_update(user_name='root',password='123456',db='learn',url='localhost'):
    # 创建连接
    conn = pymysql.connect(host=url,user=user_name,password=password,database=db)
    
    # 自动提交,默认是Flase
    print(conn.get_autocommit())
    # 创建光标
    cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
    cur_update = conn.cursor()
    
    # 更新前的数据
    cur.execute('select * from b_table where id<=11')
    print('更新前的数据：')
    rs = cur.fetchall()
    for r in rs:
        print(r)
    
    # 1--单个数据更新 execute
    try:
        sql = 'update b_table set item=%s where id=%s'
        line = ('00',0) 
        cur_update.execute(sql,line)
        # 提交
        conn.commit()
    except:
        print(cur_update.mogrify(sql,line),' 更新失败')
        # 回滚
        conn.rollback()

    # 2--多个数据更新 executemany
    # 需要更新的id列表
    id_list = (1,2,3,4,5,6,7,8,9,9,10,11)
    item_list = ('11','22','33','44','55','66','77','88','99','999','1010','1111')
    lines = []
    
    try:
        sql = 'update b_table set item=%s where id=%s'
        # 生成更新列表
        for id,item in zip(id_list,item_list):
            lines.append((item,id))
            
        print(lines)
        # 运行多个更新
        cur_update.executemany(sql,lines)
    
        conn.commit()
    except:
        print(cur_update.mogrify(sql,lines[0]),' 更新失败')
        conn.rollback()
    
    # 更新后的数据
    cur.execute('select * from b_table where id<=11')
    print('更新后的数据：')
    rs = cur.fetchall()
    for r in rs:
        print(r)

    # 关闭光标
    cur_update.close()
    cur.close()
    # 关闭连接
    conn.close()

2.2.5 executemany与execute及执行效率讨论

在写cx_oracle的那一篇的批处理语句时，有提到过cx_oracle提供的获取executemany运行结果的方法，因为executemany不会抛出异常。

而在pymysql中，的确也有这种情况：也就是说，executemany中有一条或者全部输入报错，并不会跳转到异常处理（上面的异常处理语句是摆设，没有用）。executemany会返回运行成功的语句数量，仅此而已。

我查看了下help(pymysql.cursors.Cursor)下并不包含相关方法来获取错误信息。

我又使用help查看了一下executemany的定义，只是最后一句“这个方法提升了多行插入和替换的性能。否则它跟循环执行execute()差不多”。

我感觉不同方法运行同样规模的数据库语句语句运行的效率在一般情况下是差不多的，区别在于提交commi的时机。如果不进行提交的话，数据占据着内存，如果操作大量数据的时候频繁提交，则会拉低执行效率，需要自己进行取舍。

所以我感觉executemany和execute的底层是一致的，在使用合适地方法调用execute运行循环的时候其执行效率与executemany是一致的。如果的确找不到executemany数据出错时报错的返回，应该会对两种方式的使用进行取舍。

 |  executemany(self, query, args)
 |      Run several data against one query
 |
 |      :param query: query to execute on server
 |      :param args:  Sequence of sequences or mappings.  It is used as parameter.
 |      :return: Number of rows affected, if any.
 |
 |      This method improves performance on multiple-row INSERT and
 |      REPLACE. Otherwise it is equivalent to looping over args with
 |      execute().

3、实践：多线程读写数据库

实现多线程对同一张表查询数据并拼接结果，将结果写入同一个文件内

3.1 多线程进行数据库操作

一般提供的数据库操作对象会包含连接对象和光标对象。在多线程进行操作的时候，需要考虑以下哪种方式比较合适：

多线程访问同一个cursor
多线程访问同一个连接connect
多线程创建不同的连接connect

第一种共享同一个cursor的时候还需要进行加锁，否则会报错，其实际上在数据库访问中还是进行的单线程访问。第二种从一个连接创建多个cursor交给多线程使用，我查到的说法是并不推荐，好像会有约束性的问题？所以一般推荐第三种方法，即创建多个连接进行数据库访问，再延伸一下可以创建连接池方便使用。

3.2 表分页处理

既然要分多个连接访问数据库，要怎么保证多线程查询到的数据无重复并且能够完整获取？一般考虑使用分页处理，直接从SQL语句中进行处理。mysql和oracle不同，支持limit语句。

select * from b_table limit 100,20

3.3 数据库连接池

面对大量的web请求和插入与查询请求，mysql连接会不稳定，出现错误’Lost connection to MySQL server during query ([Errno 104] Connection reset by peer)’。

而且原理跟线程池相似，多线程频繁创建和关闭数据库连接会出现许多不必要的开销，所以维持一个连接池是一个很好的选择。

DBUtils是Python的一个用于实现数据库连接池的模块。

使用pip install安装模块后导入使用。

PooledDB 提供线程间可共享的数据库连接，把连接放回连接池而不是真正的关闭，可重用。

PooledDB(creator,  # 使用链接数据库的模块
         mincached=0, # 最少的空闲连接数
         maxcached=0, # 最大的空闲链接
         maxshared=0, #当连接数达到这个数时,新请求的连接会分享已经分配出去的连接
         maxconnections=0,# 最大的连接数
         blocking=False, #当这个为True时，查过最大连接数后新的请求会等待
         maxusage=None, 
         setsession=None, 
         reset=True, 
         failures=None, 
         ping=1, 
         *args, **kwargs)

3.4 多线程访问同一个文件

之前考虑过多线程写入不同文件后拼接。但好像，至少对于python来说这样子多线程就无意义了。我这里对写入行的前后并无要求，可以使用加锁的方法进行文件追加写入。好像也有方法指定写入位置，只是没有深入了解。

    # 写法一：并发写入指定文件并加锁:追加
    def write1(self,text):
        self.lock.acquire() #加锁
        with open(self.write_file,'a+') as fo:
            fo.write(text+'
')
        self.lock.release()
        
     # 写法二（错误写法）：并发写入指定文件并加锁:追加
    def write2(self,text):
        with open(self.write_file,'a+') as fo:
            self.lock.acquire() #加锁
            fo.write(text+'
')
            self.lock.release()

上面的两种写法，我一开始采用的是第一种，结果文件中数据量对不上。我在排查问题的过程中，看到了有部分行只有半截，最终结果也是比预期结果少，所以换用了写法一。

是否是因为方法二代码中进行了打开文件后，就相当于我们已经进行了文件打开，即使加锁等待，没有马上写入。但其实在等待其他线程写入完成并保存，该等待的线程拿到写权限的那一刻，还是往旧的文件写入，而不是在上一个写入线程的文件基础上追加，造成了内容丢失。

所以说，方法二在结果上无异于没加锁。

3.5 主线程获取多线程返回数据：Queue的使用

python中多线程向主线程传回数据的方式，可以通过队列插入数据的方式。数据插入过程中应该是不需要加锁的，在其介绍中说 这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程间的同步。但是我还是加了锁，因为前面文件加锁写顺手了。

        # 结果队列
        self.res = queue.Queue(10)
        
        # 队列锁
        self.qlock = threading.Lock()
    # 将运行结果写入队列
    def write_res(self,begin,num,c):
        res = '线程【{},{}】运行结束，写入总数：{}，结束时间：{}'.format(begin,num,c,datetime.datetime.now().strftime('%Y%m%d%H%M%S'))
        
        self.qlock.acquire()
        self.res.put(res)
        self.qlock.release()

3.6 代码

import threading
import pymysql
from dbutils.pooled_db import PooledDB
import datetime,time,math
import queue

#python 关于多进程与多线程且写入同一文件情况 http://sunsunsir.cn/detail/6

class processing:

    def __init__(self,write_file,host='localhost',user_name='root',password='123456',db='learn',maxconnections=5,thread_num=5):
        # 创建数据库连接池
        self.pool = PooledDB(creator = pymysql, maxconnections=maxconnections,maxshared=maxconnections, host=host, user=user_name,
                passwd=password, db=db, port=3306, charset="utf8")
        # 线程数    
        self.thread_num = thread_num
        
        # 写文件
        self.write_file = write_file
        
        # 锁
        self.lock = threading.Lock()
        
        # 结果队列
        self.res = queue.Queue(10)
        
        # 队列锁
        self.qlock = threading.Lock()

    

    # 每个线程运行:从数据库读取分页数据，对每条数据进行加工，写入同一个文件
    # begin,num 分页
    def thread_doing(self,begin,num):
        conn = self.pool.connection()  
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        cursor.execute('select * from b_table limit %s,%s',[begin,num])
        c = 0
        while(1):
            rs = cursor.fetchone()
            if rs == None:
                break
            text = '[{},{}]id:{},item:{},time:{},nowtime:{}'.format(begin,num,
                                rs['id'],rs['item'],rs['time'],
                    datetime.datetime.now().strftime('%Y%m%d%H%M%S'))
            self.write(text)
            c = c+1
        
        self.write_res(begin,num,c)
        cursor.close()
        conn.close()  # 将连接放回连接池
        
    # 并发写入指定文件并加锁:追加
    def write(self,text):
        self.lock.acquire() #加锁
        with open(self.write_file,'a+') as fo:
            fo.write(text+'
')
        self.lock.release()
            
    # 将运行结果写入队列
    def write_res(self,begin,num,c):
        res = '线程【{},{}】运行结束，写入总数：{}，结束时间：{}'.format(begin,num,c,datetime.datetime.now().strftime('%Y%m%d%H%M%S'))
        
        self.qlock.acquire()
        self.res.put(res)
        self.qlock.release()
            
    def test(self):
        start_time = datetime.datetime.now()
        print('开始时间：',start_time.strftime('%Y%m%d%H%M%S'))
        # 查找表中全部数据量
        conn = self.pool.connection()  
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        cursor.execute('select * from b_table limit 0,10')
        while(1):
            rs = cursor.fetchone()
            if rs==None:
                break
            print(rs)
        cursor.close()
        conn.close()
        end_time = datetime.datetime.now() 
        print('{} 完成！耗时：{} '.format(end_time.strftime('%Y%m%d%H%M%S'),end_time-start_time))
            
    def run(self):
        start_time = datetime.datetime.now()
        print('开始时间：',start_time.strftime('%Y%m%d%H%M%S'))
        # 查找表中全部数据量
        conn = self.pool.connection()  
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        cursor.execute('select count(*) count from b_table')
        count = cursor.fetchone()['count']
        cursor.close()
        conn.close()
        # 分页，向上取整
        page = math.ceil(count/self.thread_num)
        print('表数据量：{}，线程数：{}，分页大小：{}'.format(count,self.thread_num,page))
        
        # 清空文件
        with open(self.write_file, 'w') as fo:
            fo.seek(0)
            fo.truncate()
        
        # 多线程
        ths = []
        # 创建线程
        for i in range(self.thread_num):
            #print(page*i,',',page)
            ths.append(threading.Thread(target=self.thread_doing,args=(page*i,page,)))

        # 启动线程
        for i in range(self.thread_num):
            ths[i].start()
        print('等待中........')           
        # 等待线程完成
        for i in range(self.thread_num):
            ths[i].join()
            

        end_time = datetime.datetime.now() 
        print('{} 完成！耗时：{} '.format(end_time.strftime('%Y%m%d%H%M%S'),end_time-start_time))
        
        while not self.res.empty():
            print(self.res.get())
        
if __name__=='__main__':
    p = processing('a.txt')
    #p.test()
    p.run()

processing

3.7 结果

开始时间： 20211004141329
表数据量：1000057，线程数：5，分页大小：200012
等待中........
20211004141919 完成！耗时：0:05:50.015016
线程【800048,200012】运行结束，写入总数：200009，结束时间：20211004141917
线程【0,200012】运行结束，写入总数：200012，结束时间：20211004141918
线程【200012,200012】运行结束，写入总数：200012，结束时间：20211004141918
线程【400024,200012】运行结束，写入总数：200012，结束时间：20211004141919
线程【600036,200012】运行结束，写入总数：200012，结束时间：20211004141919

4、总结

4.1 模块导入的一些问题

就一些小问题。

在数据库连接涉及两个模块PyMysql和DBUtils，这个两个模块在pip list和import模块名的时候，名称大小写是不一致的，这个给使用过程中造成了些许不便。有说明是版本原因造成的。我想这个有机会再进行后续了解吧。

然后我又发现了一个奇怪的东西，仅仅记录下来先吧。

import pymysql
#from dbutils.pooled_db import PooledDB
import dbutils


if __name__=='__main__':
    print(dir(dbutils))

上面的语句注不注释“from dbutils.pooled_db import PooledDB”这一句，结果是不一样的。

## 注释第二句
['__all__', '__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__path__', '__spec__', '__version__']

## 不注释第二句
['__all__', '__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__path__', '__spec__', '__version__', 'pooled_db', 'steady_db']