这么一搞，再也不怕线程打架了

三郞 2023-09-04 发布于重庆

展开全文

这么一搞，再也不怕线程打架了_多线程

文 | 李晓飞

假如我们需要处理一个文本文件，里面有 100万行数据，需要对每条数据做处理，比如将每行数据的数字做一个运算，放入到另一个文件里。

最简单的办法就是打开文件，逐行读取，每读取一行，对这一行做下处理，添加到目标文件中，再回来读取下一行。

这就是线性处理方式，假如处理一行数据需要 0.1 秒，那么用线性处理方式就需要：

10万秒，即大概 28个小时

显然对我们来说，这个时间有点长，有没用办法缩短呢？

当然有办法，那就是用 多线程 处理！

为什么呢？是因为多线程是提高效率，实现更有效程序的必然状态。

比如，需要处理大量的数据，需要响应多样的请求，需要与慢速的处理过程交互等等，都需要用到线程编程。

但是，线程这个概念不太好理解，用起来也总是不方便，而且容易出错，一方面是因为，我们的思路是线性的，另一方面是多线程本身有很多需要掌握的概念，学习理解难度比较高。

今天我将分享一下我在工作中是如何利用多线程技术，提速增效的。

对于前面那个例子，可以将原来的一个处理流程，分解为多个，例如之前的处理可以分解为：

读取行、做运算、存文件三个自流程。

这样的话，相当于将只能一个人做的工作，可以让更多的人来做，从而形成类似的流水线效应，如图所示：

这么一搞，再也不怕线程打架了_多线程_03

流水线

这是一张 CPU 处理指令的流水线示意图，可以看到在 t3 和 t4 的时间，四个工作在同时进行。

那么用多线程，就可以使我们的三个工作出现同时运行的状态，提升效率，比如先读取一行，然后再处理数据的同时，读取下一行，如此往复。

是不是感觉很好？

先别着急，首先需要解决一个问题 ——

如何避免重复读和跳读

重复读指的是，一个以上线程读取到了同一条数据；

跳读指的是，有些数据行没有任何线程处理。

这里介绍一个帮助我处理了很多多线程问题的方法，一个数据源类。

多线程数据源类

数据源类，就是将数据集中管理，然后以线程安全的方式为多线程程序提供数据。

注意：并非最佳方法，但很实用

废话不多说，直接看代码：

import threading

class DataSource:
    def __init__(self, dataFileName, startLine=0, maxcount=None):
        self.dataFileName = dataFileName
        self.startLine = startLine  # 第一行行号为1
        self.line_index = startLine # 当前读取位置
        self.maxcount = maxcount  # 读取最大行数
        self.lock = threading.RLock() # 同步锁        

        self.__data__ = open(self.dataFileName, 'r', encoding= 'utf-8')
        for i in range(self.startLine):
            l = self.__data__.readline()

    def getLine(self):
        self.lock.acquire()
        try:
            if self.maxcount is None or self.line_index < (self.startLine + self.maxcount):
line = self.__data__.readline()
if line:
self.line_index += 1
return True, line
else:
return False, None
else:
return False, None

except Exception as e:
return False, "处理出错:" + e.args
finally:
self.lock.release()

def __del__(self):
if not self.__data__.closed:
self.__data__.close()
print("关闭数据源:", self.dataFileName)
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.

__init__ 初始化方法，接受 3 个参数
lock 属性是一个同步锁，以便在多线程读取不出现冲突

dataFileName 是数据文件路径
startLine 开始读取行，对于大文件需要分配处理时特别有用，
maxcount 读取最大行数，通过和 startLine 配合可以读取指定部分的数据，默认为全部读取

getLine 方法，每次调用会返回一个元组，包含状态和得到的，数据
__del__ 方法会在对象销毁时调用，在此记录当前处理位置

这样就是可以应用在多线程程序中，承担读取待处理记录的任务了。

业务处理

例如核心处理程序如下：

import time

def process(worker_id, datasource):
    count = 0
    while True:
        status, data = datasource.getLine()
        if status:
            print(">>> 线程[%d] 获得数据， 正在处理……" % worker_id)
            time.sleep(3) # 等待3秒模拟处理过程
            print(">>> 线程[%d] 处理数据 完成" % worker_id)
            count += 1
        else:
            break # 退出循环
    print(">>> 线程[%d] 结束， 共处理[%d]条数据" % (worker_id, count))
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.

参数 worker_id 是线程号，用于区分输出消息
参数 datasource 是 DataSource 的实例，作为各线程的共享数据源
count 用于记录当前线程处理的记录数
用一个死循环，驱动反复处理，直到读取没数据可读

组装

线程组装部分就也很简单：

import threading

def main():
    datasource = DataSource('data.txt') 
    workercount = 10 # 开启的线程数，注意：并非越多越快哦
    workers = []
    for i in range(workercount):
        worker = threading.Thread(target=process, args=(i+1, datasource))
        worker.start()
        workers.append(worker)

    for worker in workers:
        worker.join()
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.

先初始化一个 DataSource
workercount 为需要创建的线程数，在实际应用中可以通过配置或者参数提供，另外不是线程越多越好，一般设置为CPU核心数的两倍即可
threading.Thread 是线程类，可以实例化一个线程，target 参数是线程处理方法，这里就是前面定义的 process 方法，args 为提供给处理方法的参数
线程的 start 方法是启动线程，因为创建不等于启动，start 是个异步方法，调用会瞬间完成
join 方法是等待线程处理完成，是同步方法，只有线程真正处理完成才会结束