致敬白衣天使,学习Python读取
创始人
2024-05-25 07:28:18
0

名字:阿玥的小东东

学习:Python、c++

主页:阿玥的小东东

故事设定:现在学校要求对所有同学进行核酸采集,每位同学先在宿舍内等候防护人员(以下简称“大白”)叫号,叫到自己时去停车场排队等候大白对自己进行采集,采集完之后的样本由大白统一有序收集并储存。

名词解释:

  • 学生:所有的学生是一个大文件,每个学生是其中的一行数据
  • 宿舍:硬盘
  • 停车场:内存
  • 核酸采集:数据处理
  • 样本:处理后的数据
  • 大白:程序

学生数量特别少的情况

当学生数量特别少时,可以考虑将所有学生统一叫到停车场等候,再依次进行核酸采集。

方法一:简单情况

此时的程序可以模拟为:

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

import time

from typing import List

  

  

def pick_all_students(dorm: str) -> List[str]:

    with open(dorm, "rt", encoding="utf8") as fin:

        students = fin.readlines()

        return students

  

  

def pick_sample(student: str) -> str:

    time.sleep(0.01)

    sample = f"{student.strip()}'s sample"

    return sample

  

  

def process(dorm: str, sample_storeroom: str) -> None:

    with open(sample_storeroom, "wt", encoding="utf8") as fout:

        students = pick_all_students(dorm)

        for student in students:

            sample = pick_sample(student)

            fout.write(f"{sample}\n")

            fout.flush()

  

  

if __name__ == "__main__":

    process(

        "student_names.txt",

        "sample_storeroom.txt"

    )

注意,在第19行中,大白一次性把所有同学都叫到了停车场中。这种做法在学生比较少时做起来很快,但是如果学生特别多,停车场装不下怎么办?

停车场空间不够时怎么办?

方法二:边读边处理

一般来说,由于停车场空间有限,我们不会采用一次性把所有学生都叫到停车场中,而是会一个一个地处理,这样可以节约内存空间。

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

import time

from typing import Iterator

  

  

def pick_one_student(dorm: str) -> Iterator[str]:

    with open(dorm, "rt", encoding="utf8") as fin:

        for student in fin:

            yield student

  

  

def pick_sample(student: str) -> str:

    time.sleep(0.01)

    sample = f"{student.strip()}'s sample"

    return sample

  

  

def process(dorm: str, sample_storeroom: str) -> None:

    with open(sample_storeroom, "wt", encoding="utf8") as fout:

        for student in pick_one_student(dorm):

            sample = pick_sample(student)

            fout.write(f"{sample}\n")

            fout.flush()

  

  

if __name__ == "__main__":

    process(

        "student_names.txt",

        "sample_storeroom.txt"

    )

这里pick_one_student函数中的返回值是用yield返回的,一次只会返回一名同学。

不过,这种做法虽然确保了停车场不会满员,但是这种做法在人数特别多的时候就不再适合了。虽然可以保证完成任务,但由于每次只能采集一个同学,程序的执行并不高。特别是当你的CPU有多个核时,会浪费机器性能,出现一核有难,其它围观的现象。

怎么加快执行效率?

大家可能也已经注意到了,刚刚我们的场景中,不论采用哪种方法,都只有一名大白在工作。那我们能不能加派人手,从而提高效率呢?

答案当然是可行的。我们现在先考虑增加两名大白,使得一名大白专注于叫号,安排学生进入停车场,另外一名大白专注于采集核酸,最后一名大白用于存储核酸样本。

方法三

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

import time

from multiprocessing import Queue, Process

from typing import Iterator

  

  

def pick_student(stu_queue: Queue, dorm: str) -> Iterator[str]:

    print("pick_student: started")

  

    picked_num = 0

    with open(dorm, "rt", encoding="utf8") as fin:

        for student in fin:

            stu_queue.put(student)

            picked_num += 1

            if picked_num % 500 == 0:

                print(f"pick_student: {picked_num}")

  

    # end signal

    stu_queue.put(None)

    print("pick_student: finished")

  

  

def pick_sample(student: str) -> str:

    time.sleep(0.01)

    sample = f"{student.strip()}'s sample"

    return sample

  

  

def process(stu_queue: Queue, store_queue: Queue) -> None:

    print("process: started")

  

    process_num = 0

    while True:

        student = stu_queue.get()

        if student is not None:

            sample = pick_sample(student)

            store_queue.put(sample)

            process_num += 1

            if process_num % 500 == 0:

                print(f"process: {process_num}")

        else:

            break

  

    # end signal

    store_queue.put(None)

    print("process: finished")

  

  

def store_sample(store_queue: Queue, sample_storeroom: str) -> None:

    print("store_sample: started")

  

    store_num = 0

    with open(sample_storeroom, "wt", encoding="utf8") as fout:

        while True:

            sample = store_queue.get()

            if sample is not None:

                fout.write(f"{sample}\n")

                fout.flush()

  

                store_num += 1

                if store_num % 500 == 0:

                    print(f"store_sample: {store_num}")

            else:

                break

  

    print("store_sample: finished")

  

  

if __name__ == "__main__":

    dorm = "student_names.txt"

    sample_storeroom = "sample_storeroom.txt"

  

    stu_queue = Queue()

    store_queue = Queue()

  

    store_p = Process(target=store_sample, args=(store_queue, sample_storeroom), daemon=True)

    store_p.start()

    process_p = Process(target=process, args=(stu_queue, store_queue), daemon=True)

    process_p.start()

    read_p = Process(target=pick_student, args=(stu_queue, dorm), daemon=True)

    read_p.start()

  

    store_p.join()

这份代码中,我们引入了多进程的思路,将每个大白看作一个进程,并使用了队列Queue作为进程间通信的媒介。stu_queue表示学生叫号进停车场的队列,store_queue表示已经采集过的待存储核酸样本的队列。

此外,为了控制进程的停止,我们在pick_student和 process函数的最后都向各自队列中添加了None作为结束标志符。

假设有1w名学生(student_names.txt文件有1w行),经过测试后发现上述方法的时间如下:

  • 方法一:1m40.716s
  • 方法二:1m40.717s
  • 方法三:1m41.097s

咦?不是做了分工吗?怎么速度还变慢了?经笔者观察,这是因为叫号的大白速度太快了(文件读取速度快)通常是TA已经齐活了,另外俩人还在吭哧吭哧干活呢,体现不出来分工的优势。如果这个时候我们对法二和法三的叫号做延时操作,每个学生叫号之后停滞10ms再叫下一位学生,则方法三的处理时间几乎不变,而方法二的时间则会延长至3m21.345s。

怎么加快处理速度?

上面提到,大白采核酸的时间较长,往往上一个人的核酸还没采完,下一个人就已经在后面等着了。我们能不能提高核酸采集这个动作(数据处理)的速度呢?其实一名大白执行一次核酸采集的时间我们几乎无法再缩短了,但是我们可以通过增加人手的方式,来达到这个目的。就像去银行办业务,如果开放的窗口越多,那么每个人等待的时间就会越短。这里我们也采取类似的策略,增加核酸采集的窗口。

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

import time

from multiprocessing import Queue, Process, cpu_count

from typing import Iterator

  

  

def pick_student(stu_queue: Queue, dorm: str, num_workers: int) -> Iterator[str]:

    print("pick_student: started")

  

    picked_num = 0

    with open(dorm, "rt", encoding="utf8") as fin:

        for student in fin:

            stu_queue.put(student)

            picked_num += 1

            if picked_num % 500 == 0:

                print(f"pick_student: {picked_num}")

  

    # end signal

    for _ in range(num_workers):

        stu_queue.put(None)

  

    print("pick_student: finished")

  

  

def pick_sample(student: str) -> str:

    time.sleep(0.01)

    sample = f"{student.strip()}'s sample"

    return sample

  

  

def process(stu_queue: Queue, store_queue: Queue) -> None:

    print("process: started")

  

    process_num = 0

    while True:

        student = stu_queue.get()

        if student is not None:

            sample = pick_sample(student)

            store_queue.put(sample)

            process_num += 1

            if process_num % 500 == 0:

                print(f"process: {process_num}")

        else:

            break

  

    print("process: finished")

  

  

def store_sample(store_queue: Queue, sample_storeroom: str) -> None:

    print("store_sample: started")

  

    store_num = 0

    with open(sample_storeroom, "wt", encoding="utf8") as fout:

        while True:

            sample = store_queue.get()

            if sample is not None:

                fout.write(f"{sample}\n")

                fout.flush()

  

                store_num += 1

                if store_num % 500 == 0:

                    print(f"store_sample: {store_num}")

            else:

                break

  

    print("store_sample: finished")

  

  

if __name__ == "__main__":

    dorm = "student_names.txt"

    sample_storeroom = "sample_storeroom.txt"

    num_process = max(1, cpu_count() - 1)

  

    maxsize = 10 * num_process

    stu_queue = Queue(maxsize=maxsize)

    store_queue = Queue(maxsize=maxsize)

  

    store_p = Process(target=store_sample, args=(store_queue, sample_storeroom), daemon=True)

    store_p.start()

    process_workers = []

    for _ in range(num_process):

        process_p = Process(target=process, args=(stu_queue, store_queue), daemon=True)

        process_p.start()

        process_workers.append(process_p)

    read_p = Process(target=pick_student, args=(stu_queue, dorm, num_process), daemon=True)

    read_p.start()

  

    for worker in process_workers:

        worker.join()

  

    # end signal

    store_queue.put(None)

    store_p.join()

总耗时 0m4.160s !我们来具体看看其中的细节部分:

首先我们将CPU核数 - 3作为采核酸的大白数量。这里减3是为其它工作进程保留了一些资源,你也可以根据自己的具体情况做调整

这次我们在 Queue中增加了 maxsize参数,这个参数是限制队列的最大长度,这个参数通常与你的实际内存情况有关。如果数据特别多时要考虑做些调整。这里我采用10倍的工作进程数目作为队列的长度

注意这里pick_student函数中要为每个后续的工作进程都添加一个结束标志,因此最后会有个for循环

我们把之前放在process函数中的结束标志提取出来,放在了最外侧,使得所有工作进程均结束之后再关闭最后的store_p进程

结语

总结来说,如果你的数据集特别小,用法一;通常情况下用法二;数据集特别大时用法四。

相关内容

热门资讯

喜欢穿一身黑的男生性格(喜欢穿... 今天百科达人给各位分享喜欢穿一身黑的男生性格的知识,其中也会对喜欢穿一身黑衣服的男人人好相处吗进行解...
发春是什么意思(思春和发春是什... 本篇文章极速百科给大家谈谈发春是什么意思,以及思春和发春是什么意思对应的知识点,希望对各位有所帮助,...
网络用语zl是什么意思(zl是... 今天给各位分享网络用语zl是什么意思的知识,其中也会对zl是啥意思是什么网络用语进行解释,如果能碰巧...
为什么酷狗音乐自己唱的歌不能下... 本篇文章极速百科小编给大家谈谈为什么酷狗音乐自己唱的歌不能下载到本地?,以及为什么酷狗下载的歌曲不是...
华为下载未安装的文件去哪找(华... 今天百科达人给各位分享华为下载未安装的文件去哪找的知识,其中也会对华为下载未安装的文件去哪找到进行解...
怎么往应用助手里添加应用(应用... 今天百科达人给各位分享怎么往应用助手里添加应用的知识,其中也会对应用助手怎么添加微信进行解释,如果能...
家里可以做假山养金鱼吗(假山能... 今天百科达人给各位分享家里可以做假山养金鱼吗的知识,其中也会对假山能放鱼缸里吗进行解释,如果能碰巧解...
四分五裂是什么生肖什么动物(四... 本篇文章极速百科小编给大家谈谈四分五裂是什么生肖什么动物,以及四分五裂打一生肖是什么对应的知识点,希...
一帆风顺二龙腾飞三阳开泰祝福语... 本篇文章极速百科给大家谈谈一帆风顺二龙腾飞三阳开泰祝福语,以及一帆风顺二龙腾飞三阳开泰祝福语结婚对应...
美团联名卡审核成功待激活(美团... 今天百科达人给各位分享美团联名卡审核成功待激活的知识,其中也会对美团联名卡审核未通过进行解释,如果能...