48.线程

线程概念的引入背景

进程

之前我们已经了解了操作系统中进程的概念，程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。

程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。

在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行。这是这样的设计，大大提高了CPU的利用率。

进程的出现让每个用户感觉到自己独享CPU，因此，进程就是为了在CPU上实现多道编程而提出的。

1、进程（有时被称为重量级进程）是程序的一次执行。每个进程都有自己的地址空间，内存，数据栈以及其它记录其运行轨迹的辅助数据
2、操作系统管理在其上运行的所有进程，并为这些进程公平地分配时间
3、进程也可以通过 fork 和 spawn 操作来完成其它的任务
4、不过各个进程有自己的内存空间，数据栈等，所以只能使用进程间通讯（IPC），而不能直接共享信息

 

有了进程为什么要有线程

　进程有很多优点，它提供了多道编程，让我们感觉我们每个人都拥有自己的CPU和其他资源，可以提高计算机的利用率。

很多人就不理解了，既然进程这么优秀，为什么还要线程呢？其实，仔细观察就会发现进程还是有很多缺陷的，主要体现在两点上：

    进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。

    进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起，即使进程中有些工作不依赖于输入的数据，也将无法执行。

如果这两个缺点理解比较困难的话，举个现实的例子也许你就清楚了：

如果把我们上课的过程看成一个进程的话，那么我们要做的是耳朵听老师讲课，手上还要记笔记，脑子还要思考问题，这样才能高效的完成听课的任务。

而如果只提供进程这个机制的话，上面这三件事将不能同时执行，同一时间只能做一件事，听的时候就不能记笔记，也不能用脑子思考，这是其一；

如果老师在黑板上写演算过程，我们开始记笔记，而老师突然有一步推不下去了，阻塞住了，他在那边思考着，

而我们呢，也不能干其他事，即使你想趁此时思考一下刚才没听懂的一个问题都不行，这是其二。

现在你应该明白了进程的缺陷了，而解决的办法很简单，我们完全可以让听、写、思三个独立的过程，并行起来，这样很明显可以提高听课的效率。而实际的操作系统中，也同样引入了这种类似的机制——线程。

线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务

 什么是线程

1.线程（有时被称为轻量级进程）跟进程有些相似，不同的是，所有的线程运行在同一个进程中，共享相同的运行环境 。

2.它们可以想像成是在主进程或“主线程”中并行运行的“迷你进程”。
3.线程有开始，顺序执行和结束三部分。它有一个自己的指令指针，记录自己运行到什么地方。线程的运行可能被抢占（中断），或暂时的被挂起（也叫睡眠），让其它的线程运行，这叫做让步
4.一个进程中的各个线程之间共享同一片数据空间，所以线程之间可以比进程之间更方便地共享数据以及相互通讯
5.线程一般都是并发执行的，正是由于这种并行和数据共享的机制使得多个任务的合作变为可能
6.实际上，在单 CPU 的系统中，真正的并发是不可能的，每个线程会被安排成每次只运行一小会，然后就把 CPU 让出来，让其它的线程去运行。
7.在进程的整个运行过程中，每个线程都只做自己的事，在需要的时候跟其它的线程共享运行的结果
8.当然，这样的共享并不是完全没有危险的。如果多个线程共同访问同一片数据，则由于数据访问的顺序不一样，有可能导致数据结果的不一致的问题。这叫做竞态条件(race condition)
9.幸运的是，大多数线程库都带有一系列的同步原语，来控制线程的执行和数据的访问
10.另一个要注意的地方是，由于有的函数会在完成之前阻塞住，在没有特别为多线程做修改的情况下，这种“贪婪”的函数会让 CPU 的时间分配有所倾斜。导致各个线程分配到的运行时间可能不尽相同，不尽公平

线程的出现

　　
60年代，在OS中能拥有资源和独立运行的基本单位是进程，然而随着计算机技术的发展，进程出现了很多弊端，一是由于进程是资源拥有者，创建、撤消与切换存在较大的时空开销，因此需要引入轻型进程；二是由于对称多处理机（SMP）出现，可以满足多个运行单位，而多个进程并行开销过大。
　　
因此在80年代，出现了能独立运行的基本单位——线程（Threads）。
　
 
　注意：进程是资源分配的最小单位,线程是CPU调度的最小单位.
　　　　　每一个进程中至少有一个线程。　

进程和线程的关系

 如图：

　　

　　

线程和进程的区别

    1.Threads share the address space of the process that created it; processes have their own address space.
    2.Threads have direct access to the data segment of its process; processes have their own copy of the data segment of the parent process.
    3.Threads can directly communicate with other threads of its process; processes must use interprocess communication to communicate with sibling processes.
    4.New threads are easily created; new processes require duplication of the parent process.
    5.Threads can exercise considerable control over threads of the same process; processes can only exercise control over child processes.
    6.Changes to the main thread (cancellation, priority change, etc.) may affect the behavior of the other threads of the process; changes to the parent process does not affect child processes.

翻译：

1，线程共享创建它的进程的地址空间;进程有自己的地址空间。
 
2，线程可以直接访问其进程的数据段;进程有自己的父进程数据段的副本。
 
3，线程可以直接与进程的其他线程通信;进程必须使用进程间通信来与同胞进程通信。
 
4，新线程很容易创建;新进程需要父进程的重复。
 
5，线程可以对相同进程的线程进行相当大的控制;进程只能对子进程进行控制。
 
6，对主线程的更改(取消、优先级更改等)可能会影响进程的其他线程的行为;对父进程的更改不会影响子进程。

总结上述区别，无非两个关键点，这也是我们在特定的场景下需要使用多线程的原因：

   1. 同一个进程内的多个线程共享该进程内的地址资源
   2. 创建线程的开销要远小于创建进程的开销（创建一个进程，就是创建一个车间，涉及到申请空间，而且在该空间内建至少一条流水线，但创建线程，
　　　　就只是在一个车间内造一条流水线，无需申请空间，所以创建开销小）

线程的特点

 
在多线程的操作系统中，通常是在一个进程中包括多个线程，每个线程都是作为利用CPU的基本单位，是花费最小开销的实体。线程具有以下属性。

轻型实体

　　线程中的实体基本上不拥有系统资源，只是有一点必不可少的、能保证独立运行的资源。
　　线程的实体包括程序、数据和TCB。线程是动态概念，它的动态特性由线程控制块TCB（Thread Control Block）描述。

TCB包括以下信息

 
（1）线程状态。
（2）当线程不运行时，被保存的现场资源。
（3）一组执行堆栈。
（4）存放每个线程的局部变量主存区。
（5）访问同一个进程中的主存和其它资源。
用于指示被执行指令序列的程序计数器、保留局部变量、少数状态参数和返回地址等的一组寄存器和堆栈。

独立调度和分派的基本单位

 
在多线程OS中，线程是能独立运行的基本单位，因而也是独立调度和分派的基本单位。由于线程很“轻”，故线程的切换非常迅速且开销小（在同一进程中的）。

共享进程资源

 
线程在同一进程中的各个线程，都可以共享该进程所拥有的资源，这首先表现在：所有线程都具有相同的进程id，这意味着，线程可以访问该进程的每一个内存资源；
此外，还可以访问进程所拥有的已打开文件、定时器、信号量机构等。由于同一个进程内的线程共享内存和文件，所以线程之间互相通信不必调用内核。

可并发执行

 
 

在一个进程中的多个线程之间，可以并发执行，甚至允许在一个进程中所有线程都能并发执行；同样，不同进程中的线程也能并发执行，
充分利用和发挥了处理机与外围设备并行工作的能力。

使用线程的实际场景

 
 
 

开启一个字处理软件进程，该进程肯定需要办不止一件事情，比如监听键盘输入，处理文字，定时自动将文字保存到硬盘，这三个任务操作的都是同一块数据，因而不能用多进程。只能在一个进程里并发地开启三个线程,如果是单线程，那就只能是，键盘输入时，不能处理文字和自动保存，自动保存时又不能输入和处理文字。

内存中的线程

 如图：

　
多个线程共享同一个进程的地址空间中的资源，是对一台计算机上多个进程的模拟，有时也称线程为轻量级的进程。

而对一台计算机上多个进程，则共享物理内存、磁盘、打印机等其他物理资源。多线程的运行也多进程的运行类似，是cpu在多个线程之间的快速切换。

不同的进程之间是充满敌意的，彼此是抢占、竞争cpu的关系，如同迅雷会和QQ抢资源。而同一个进程是由一个程序员的程序创建，所以同一进程内的线程是合作关系，一个线程可以访问另外一个线程的内存地址，大家都是共享的，一个线程干死了另外一个线程的内存，那纯属程序员脑子有问题。

类似于进程，每个线程也有自己的堆栈，不同于进程，线程库无法利用时钟中断强制线程让出CPU，可以调用thread_yield运行线程自动放弃cpu，让另外一个线程运行。

线程通常是有益的，但是带来了不小程序设计难度，线程的问题是：

　1. 父进程有多个线程，那么开启的子线程是否需要同样多的线程
  2.在同一个进程中，如果一个线程关闭了文件，而另外一个线程正准备往该文件内写内容呢？
因此，在多线程的代码中，需要更多的心思来设计程序的逻辑、保护程序的数据。
 
 

用户级线程和内核级线程（了解）

 
 
线程的实现可以分为两类：用户级线程(User-Level Thread)和内核线线程(Kernel-Level Thread)，后者又称为内核支持的线程或轻量级进程。在多线程操作系统中，各个系统的实现方式并不相同，在有的系统中实现了用户级线程，有的系统中实现了内核级线程。
 
 

用户级线程

内核的切换由用户态程序自己控制内核切换,不需要内核干涉，少了进出内核态的消耗，但不能很好的利用多核CPU
 
 
 如图：
 在用户空间模拟操作系统对进程的调度，来调用一个进程中的线程，每个进程中都会有一个运行时系统，用来调度线程。此时当该进程获取cpu时，进程内再调度出一个线程去执行，同一时刻只有一个线程执行。
 
 

内核级线程

 内核级线程:切换由内核控制，当线程进行切换的时候，由用户态转化为内核态。切换完毕要从内核态返回用户态；可以很好的利用smp，即利用多核cpu。windows线程就是这样的
 

用户级与内核级线程的对比

用户级线程和内核级线程的区别

1 内核支持线程是OS内核可感知的，而用户级线程是OS内核不可感知的。
2 用户级线程的创建、撤消和调度不需要OS内核的支持，是在语言（如Java）这一级处理的；而内核支持线程的创建、撤消和调度都需OS内核提供支持，
　　而且与进程的创建、撤消和调度大体是相同的。
3 用户级线程执行系统调用指令时将导致其所属进程被中断，而内核支持线程执行系统调用指令时，只导致该线程被中断。
4 在只有用户级线程的系统内，CPU调度还是以进程为单位，处于运行状态的进程中的多个线程，由用户程序控制线程的轮换运行；
　　在有内核支持线程的系统内，CPU调度则以线程为单位，由OS的线程调度程序负责线程的调度。
5 用户级线程的程序实体是运行在用户态下的程序，而内核支持线程的程序实体则是可以运行在任何状态下的程序。

内核线程的优缺点

优点：当有多个处理机时，一个进程的多个线程可以同时执行。
缺点：由内核进行调度。

用户级线程的优缺点

优点：
线程的调度不需要内核直接参与，控制简单。
可以在不支持线程的操作系统中实现。
创建和销毁线程、线程切换代价等线程管理的代价比内核线程少得多。
允许每个进程定制自己的调度算法，线程管理比较灵活。
线程能够利用的表空间和堆栈空间比内核级线程多。
同一进程中只能同时有一个线程在运行，如果有一个线程使用了系统调用而阻塞，那么整个进程都会被挂起。另外，页面失效也会产生同样的问题。
缺点：
资源调度按照进程进行，多个处理机下，同一个进程中的线程只能在同一个处理机下分时复用

混合实现

用户级与内核级的多路复用，内核同一调度内核线程，每个内核线程对应n个用户线程

linux操作系统的 NPTL

历史
在内核2.6以前的调度实体都是进程，内核并没有真正支持线程。它是能过一个系统调用clone()来实现的，这个调用创建了一份调用进程的拷贝，
跟fork()不同的是,这份进程拷贝完全共享了调用进程的地址空间。LinuxThread就是通过这个系统调用来提供线程在内核级的支持的(许多以前的线程实现都完全是在用户态，
内核根本不知道线程的存在)。非常不幸的是，这种方法有相当多的地方没有遵循POSIX标准，特别是在信号处理，调度，进程间通信原语等方面。

很显然，为了改进LinuxThread必须得到内核的支持，并且需要重写线程库。为了实现这个需求，开始有两个相互竞争的项目：
IBM启动的NGTP(Next Generation POSIX Threads)项目，以及Redhat公司的NPTL。在2003年的年中，IBM放弃了NGTP，也就是大约那时，
Redhat发布了最初的NPTL。

NPTL最开始在redhat linux 9里发布，现在从RHEL3起内核2.6起都支持NPTL，并且完全成了GNU C库的一部分。

 

设计
NPTL使用了跟LinuxThread相同的办法，在内核里面线程仍然被当作是一个进程，并且仍然使用了clone()系统调用(在NPTL库里调用)。但是，
NPTL需要内核级的特殊支持来实现，比如需要挂起然后再唤醒线程的线程同步原语futex.

NPTL也是一个1*1的线程库，就是说，当你使用pthread_create()调用创建一个线程后，在内核里就相应创建了一个调度实体，在linux里就是一个新进程，
这个方法最大可能的简化了线程的实现。

除NPTL的1*1模型外还有一个m*n模型，通常这种模型的用户线程数会比内核的调度实体多。在这种实现里，线程库本身必须去处理可能存在的调度，
这样在线程库内部的上下文切换通常都会相当的快，因为它避免了系统调用转到内核态。然而这种模型增加了线程实现的复杂性,并可能出现诸如优先级反转的问题，
此外，用户态的调度如何跟内核态的调度进行协调也是很难让人满意。


Python线程

 

退出线程

1、当一个线程结束计算，它就退出了。线程可以调用 thread.exit()之类的退出函数，也可以使用Python 退出进程的标准方法，如 sys.exit()或抛出一个 SystemExit 异常等。不过，你不可以直接"杀掉"("kill")一个线程

2、我们将要讨论两个跟线程有关的模块。这两个模块中，我们不建议使用 thread模块。

这样做有很多原因，很明显的一个原因是，当主线程退出的时候，所有其它线程没有被清除就退出了。但另一个模块 threading 就能确保所有“重要的”子线程都退出后，进程才会结束。

在 Python 中使用线程

1、想要从解释器里判断线程是否可用，只要简单的在交互式解释器里尝试导入 thread 模块就行了，只要没出现错误就表示线程可用。 

>>> import thread
>>>

2、如果你的 Python 解释器在编译时，没有打开线程支持，导入模块会失败：

>>> import thread
Traceback (innermost last): File "<stdin>", line 1, in ?
ImportError: No module named thread

这种情况下，你就要重新编译你的 Python 解释器才能使用线程

没有线程支持的情况

在单线程中顺序执行两个循环。一定要一个循环结束后，另一个才能开始。总时间是各个循环运行时间之和

 
#!/usr/bin/env python

from time import sleep, ctime

def loop0():
    print 'start loop 0 at:', ctime()
    sleep(4)
    print 'loop 0 done at:', ctime()

def loop1():
    print 'start loop 1 at:', ctime()
    sleep(2)
    print 'loop 1 done at:', ctime()

def main():
    print 'starting at:', ctime()
    loop0()
    loop1()
    print 'all DONE at:', ctime()

if __name__ == '__main__':
    main()
 

执行结果：

 
starting at:  Thu May 17 12:34:48 2018
start loop 0 at:  Thu May 17 12:35:00 2018
loop 0 done at: Thu May 17 12:35:10 2018
start loop 1 at:  Thu May 17 12:35:19 2018
loop 1 done at: Thu May 17 12:35:24 2018
all done at : Thu May 17 12:35:26 2018
 

python线程模块的选择

1.Python提供了几个用于多线程编程的模块，包括thread、threading和Queue等。

2.thread和threading模块允许程序员创建和管理线程。

3.thread模块提供了基本的线程和锁的支持，而threading提供了更高级别、功能更强的线程管理的功能。

4.Queue模块允许用户创建一个可以用于多个线程之间共享数据的队列数据结构。

5、推荐使用更高级别的threading模块

6、只建议那些有经验的专家在想访问线程的底层结构的时候，才使用thread模块

　

核心提示：避免使用 thread 模块 

避免使用thread模块，因为更高级别的threading模块更为先进，对线程的支持更为完善，而且使用thread模块里的属性有可能会与threading出现冲突；

其次低级别的thread模块的同步原语很少(实际上只有一个)，而threading模块则有很多；再者，thread模块中当主线程结束时，所有的线程都会被强制结束掉，
没有警告也不会有正常的清除工作，至少threading模块能确保重要的子线程退出后进程才退出。

thread模块不支持守护线程，当主线程退出时，所有的子线程不论它们是否还在工作，都会被强行退出。

而threading模块支持守护线程，守护线程一般是一个等待客户请求的服务器，如果没有客户提出请求它就在那等着，如果设定一个线程为守护线程，

就表示这个线程是不重要的，在进程退出的时候，不用等待这个线程退出。

三、thread 模块

这个模块只做了解即可

除了产生线程外，thread 模块也提供了基本的同步数据结构锁对象(lock object，也叫原语锁，简单锁，互斥锁，互斥量，二值信号量) 。

同步原语与线程的管理是密不可分的

常用的线程函数以及 LockType 类型的锁对象的方法

start_new_thread()函数是 thread 模块的一个关键函数，它的语法与内建的 apply()函数完全一样，其参数为：函数，函数的参数以及可选的关键字参数。不同的是，函数不是在主线程里运行，而是产生一个新的线程来运行这个函数

 thread 模块和锁对象

 
start_new_thread(function,args, kwargs=None)： 产生一个新的线程，在新线程中用指定的参数和可选的kwargs 来调用这个函数。
allocate_lock() ：分配一个 LockType 类型的锁对象
exit()： 让线程退出
LockType： 类型锁对象方法
acquire(wait=None)： 尝试获取锁对象
locked()： 如果获取了锁对象返回 True，否则返回 False
release() ：释放锁
 

实例1：使用 thread 模块mtsleep1.py

这一次使用的是 thread 模块提供的简单的多线程的机制。两个循环并发地被执行（显然，短的那个先结束）。

总的运行时间为最慢的那个线程的运行时间，而不是所有的线程的运行时间之和

 
#!/usr/bin/env python
#coding:utf-8

import thread
from time import sleep,ctime

def loop0():
    print 'start loop 0 at:', ctime()
    sleep(4)
    print 'loop 0 done at:', ctime()

def loop1():
    print 'start loop 1 at:', ctime()
    sleep(2)
    print 'loop 1 done at:', ctime()


def main():
    print "starting at:",ctime()
　　#start_new_thread(function,args, kwargs=None)： 产生一个新的线程，在新线程中用指定的参数和可选的kwargs 来调用这个函数。
    thread.start_new_thread(loop0,())
    thread.start_new_thread(loop1,())
    sleep(6)
    print "all Done at:",ctime()

if __name__ == "__main__":
    main()
 

start_new_thread()要求一定要有前两个参数。所以，就算我们想要运行的函数不要参数，我们也要传一个空的元组

执行结果：

 
starting at: Thu May 17 12:53:56 2018
start loop 0 at: start loop 1 at:Thu May 17 12:53:56 2018 
Thu May 17 12:53:56 2018
loop 1 done at: Thu May 17 12:53:58 2018
loop 0 done at: Thu May 17 12:54:00 2018
all Done at: Thu May 17 12:54:02 2018
 

实例2：使用线程和锁 (mtsleep2.py)

使用锁比 mtsleep1.py 在主线程中使用 sleep()函数更合理 。

不用为线程什么时候结束再做额外的等待。使用了锁，我们就可以在两个线程都退出后，马上退出。

 
#!/usr/bin/env python
#coding:utf-8

import thread
from time import sleep,ctime

loops = [4,2]

def loop(nloop,nsec,lock):
    print "start loop",nloop,'at:',ctime()
    sleep(nsec)
    print "loop",nloop,'done at:',ctime()
    lock.release()  #：释放锁



def main():
    print "starting at:",ctime()
    locks = []   
    nloops = range(len(loops))  #len(loops)得到的结果是2，range(len(loops))的结果是[0,1],即nloops = [0,1]

    for i in nloops:
        lock = thread.allocate_lock()   #分配一个 LockType 类型的锁对象
        lock.acquire()  #acquire(wait=None)： 尝试获取锁对象
        locks.append(lock)
　　　　 #print locks #[<thread.lock object at 0x00000000030EE0F0>, <thread.lock object at 0x00000000030EE0D0>]

    for i in nloops: #i可取的是0或1
　　　　##start_new_thread(function,args, kwargs=None)： 产生一个新的线程，在新线程中用指定的参数和可选的kwargs 来调用这个函数。
        thread.start_new_thread(loop,(i,loops[i],locks[i])) #loops[i]拿到的是4或者2；locks[i]拿到的是2个线程锁对象
    for i in nloops:
        while locks[i].locked():  #locked()： 如果获取了锁对象返回 True，否则返回 False
            pass

    print "all done at:",ctime()


if __name__ == "__main__":
    main()
 

执行结果：

 
starting at: Thu May 17 13:10:40 2018
start loop 0start loop  at:1  Thu May 17 13:10:40 2018at:
 Thu May 17 13:10:40 2018
loop 1 done at: Thu May 17 13:10:42 2018
loop 0 done at: Thu May 17 13:10:44 2018
all done at: Thu May 17 13:10:44 2018
 

提示：

 
>>> loops = [4,2]
>>> len(loops)
2
>>> range(len(loops))
[0, 1]
>>>
 

threading模块

multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性

官方链接：

https://docs.python.org/3/library/threading.html?highlight=threading#

threading 模块对象

 
Thread :表示一个线程的执行的对象
Lock :锁原语对象（跟 thread 模块里的锁对象相同）
RLock :可重入锁对象。使单线程可以再次获得已经获得了的锁（递归锁定）。
Condition: 条件变量对象能让一个线程停下来， 等待其它线程满足了某个 “条件”。如，状态的改变或值的改变。
Event: 通用的条件变量。多个线程可以等待某个事件的发生，在事件发生后，所有的线程都会被激活。
Semaphore :为等待锁的线程提供一个类似“等候室”的结构
BoundedSemaphore :与 Semaphore 类似，只是它不允许超过初始值
Timer :与 Thread 相似，只是，它要等待一段时间后才开始运行
 

threading 模块中的其它函数

除了各种同步对象和线程对象外，threading 模块还提供了一些函数 

activeCount() ：当前活动的线程对象的数量
currentThread() ：返回当前线程对象
enumerate() ：返回当前活动线程的列表
settrace(func)： 为所有线程设置一个跟踪函数
setprofile(func)： 为所有线程设置一个 profile 函数

 threading模块提供的一些方法：

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

核心提示：守护线程

 
核心提示：守护线程
1、另一个避免使用 thread 模块的原因是，它不支持守护线程。当主线程退出时，所有的子线程不论它们是否还在工作，都会被强行退出。
有时，我们并不期望这种行为，这时，就引入了守护线程的概念
2、threading 模块支持守护线程， 它们是这样工作的：
　　a.守护线程一般是一个等待客户请求的服务器，如果没有客户提出请求，它就在那等着。
　　b.如果你设定一个线程为守护线程，就表示你在说这个线程是不重要的，在进程退出的时候，不用等待这个线程退出。

3、如果你的主线程要退出的时候，不用等待那些子线程完成，那就设定这些线程的 daemon 属性。
即，在线程开始（调用 thread.start()）之前， 调用 setDaemon()函数设定线程的 daemon 标志（thread.setDaemon(True)） 就表示这个线程“不重要”
4、如果你想要等待子线程完成再退出， 那就什么都不用做，或者显式地调用thread.setDaemon(False)以保证其 daemon 标志为 False。
 你可以调用 thread.isDaemon()函数来判断其 daemon 标志的值。新的子线程会继承其父线程的 daemon 标志。
整个 Python 会在所有的非守护线程退出后才会结束,即进程中没有非守护线程存在的时候才结束。
 

Thread 类

语法：

threading.Thread(self, group=None, target=None, name=None, args=(), kwargs=None, verbose=None)

说明：

group:应为None; 在实现ThreadGroup类时保留用于将来的扩展。
target: 是run（）方法调用的可调用对象。 默认为None，表示不调用任何内容。
name:是线程名称。 默认情况下，唯一名称由“Thread-N”形式构成，其中N是小十进制数。
args:是目标调用的参数元组。 默认为（）
kwargs :是目标调用的关键字参数字典。 默认为{}。

1、threading 的 Thread 类是你主要的运行对象。它有很多 thread 模块里没有的函数
2、用 Thread 类，你可以用多种方法来创建线程。我们在这里介绍三种比较相像的方法。你可以任选一种你喜欢的，

或最适合你的程序以及最能满足程序可扩展性的（我们一般比较喜欢最后一个选择）： 

1.创建一个 Thread 的实例，传给它一个函数
2.创建一个 Thread 的实例，传给它一个可调用的类对象
3.从 Thread 派生出一个子类，创建一个这个子类的实例

Thread 对象的函数

 
start() ：开始线程的执行，线程准备就绪，等待CPU调度
run() ：定义线程的功能的函数（一般会被子类重写）。 线程被cpu调度后自动执行线程对象的run方法
join(timeout=None) ：程序挂起，直到线程结束；如果给了 timeout，则最多阻塞 timeout 秒。
　　　　　　　　　　　　逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义
getName()： 返回线程的名字。
setName(name)： 设置线程的名字。设置为后台线程或前台线程（默认）。
　　　　　　　如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止
           如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止
isAlive() ：布尔标志，表示这个线程是否还在运行中
isDaemon() ：返回线程的 daemon 标志
setDaemon(daemonic)： 把线程的 daemon 标志设为 daemonic （一定要在调用 start()函数前调用）
 

注意：

python的threading.Thread类有一个run方法，用于定义线程的功能函数，可以在自己的线程类中覆盖该方法。而创建自己的线程实例后，通过Thread类的start方法，可以启动该线程，交给python虚拟机进行调度，当该线程获得执行的机会时，就会调用run方法执行线程


开启线程的方法1：传递函数给Thread类

实例化一个 Thread（调用 Thread()）与调用 thread.start_new_thread()之间最大的区别就是，新的线程不会立即开始。

在你创建线程对象，但不想马上开始运行线程的时候，这是一个很有用的同步特性。

1、多线程编程有多种方法，传递函数给threading模块的Thread类是介绍的第一种方法

2、Thread对象使用start()方法开始线程的执行，使用join()方法挂起程序，直到线程结束

 3、threading 模块的 Thread 类有一个 join()函数，允许主线程等待线程的结束 

#!/usr/bin/env python
# coding:utf-8

import threading
import time

nums = [4, 2]


def loop(nloop, nsec):  # 定义函数，打印运行的起止时间
    print "start loop %d,at %s" % (nloop, time.ctime())
    time.sleep(nsec)
    print "loop %d done at %s " % (nloop, time.ctime())


def main():
    print "starting at: %s", time.ctime()
    threads = []

    # nloops = range(len(nums))   #nloops = [0,1]
    # for i in range(len(nums)):#创建两个线程，放入列表


    for i in range(2):  # 创建两个线程，放入列表
        t = threading.Thread(target=loop, args=(0, nums[i]))
        threads.append(t)

    for i in range(2):
        threads[i].start()  # 同时运行两个线程

    for i in range(2):
        threads[i].join()  # 主程序挂起，直到所有线程结束

    print "all Done at %s" % time.ctime()

if __name__ == "__main__":
    main()

提示：

 
所有的线程都创建了之后，再一起调用 start()函数启动，而不是创建一个启动一个。而且，不用再管理一堆锁（分配锁， 获得锁， 释放锁， 检查锁的状态等）， 
只要简单地对每个线程调用 join()函数就可以了。join()会等到线程结束，或者在给了 timeout 参数的时候，等到超时为止。
使用 join()看上去会比使用一个等待锁释放的无限循环清楚一些(这种锁也被称为"spinlock")

join()的另一个比较重要的方面是它可以完全不用调用。一旦线程启动后，就会一直运行，直到线程的函数结束，退出为止。
如果你的主线程除了等线程结束外，还有其它的事情要做（如处理或等待其它的客户请求）， 那就不用调用 join()， 
只有在你要等待线程结束的时候才要调用 join()。
 

执行结果：

 
starting at: %s Thu May 17 15:43:03 2018
start loop 0,at Thu May 17 15:43:03 2018
start loop 0,at Thu May 17 15:43:03 2018
loop 0 done at Thu May 17 15:43:05 2018 
loop 0 done at Thu May 17 15:43:07 2018 
all Done at Thu May 17 15:43:07 2018
 
例子1：
 
  
#!/usr/bin/env python
# coding:utf-8

import time
import random
from threading import Thread


def study(name):
    print("%s is learning" % name)
    time.sleep(random.randint(1, 3))
    print("%s is playing" % name)


if __name__ == '__main__':
    t = Thread(target=study, args=('james',))
    t.start()
    print("主线程开始运行....")
执行结果：
 
james is learning
主线程开始运行....
james is playing

开启线程的方法2：传递可调用类给Thread类

1、传递可调用类给Thread类是介绍的第二种方法

2、相对于一个或几个函数来说，由于类对象里可以使用类的强大的功能，可以保存更多的信息，这种方法更为灵活

 
#!/usr/bin/env python
#coding:utf-8

import threading
import time

nums = [4,2]

class ThreadFunc(object):#定义可调用的类
    def __init__(self,func,args,name=''):
        self.name = name
        self.func = func
        self.args = args

    def __call__(self):
        apply(self.func,self.args)

def loop(nloop,nsec): #定义函数，打印运行的起止时间
    print "start loop %d,at %s" % (nloop,time.ctime())
    time.sleep(nsec)
    print "loop %d done at %s " % (nloop,time.ctime())

def main():
     print 'starting at: %s' % time.ctime()
     threads = []

     for i in range(2):
         t = threading.Thread(target = ThreadFunc(loop, (i, nums[i]),loop.__name__))
         threads.append(t) #创建两个线程，放入列表

     for i in range(2):
         threads[i].start()

     for i in range(2):
         threads[i].join()
         print 'all Done at %s' % time.ctime()

if __name__ == '__main__':
    main()
 

执行结果：

 
starting at: Thu May 17 16:05:12 2018
start loop 0,at Thu May 17 16:05:12 2018
start loop 1,at Thu May 17 16:05:12 2018
loop 1 done at Thu May 17 16:05:14 2018 
loop 0 done at Thu May 17 16:05:16 2018 
all Done at Thu May 17 16:05:16 2018
all Done at Thu May 17 16:05:16 2018
 

例子：

#!/usr/bin/env python
# coding:utf-8



from threading import Thread
import time


class MyThread(Thread):
    def __init__(self, name):
        #super().__init__()  #python 3
        super(MyThread,self).__init__()  #python 2
        self.name = name

    def run(self):
        print('%s is learning' % self.name)
        time.sleep(2)
        print('%s is playing' % self.name)


if __name__ == '__main__':
    t1 = MyThread('james')
    t1.start()
    print("主线程开始运行....")

开启线程的方法3:从 Thread 派生出一个子类，创建一个这个子类的实例

我们现在要子类化 Thread 类，而不是创建它的实例。这样做可以更灵活地定制我们的线程对象，而且在创建线程的时候也更简单。 

#!/usr/bin/env python
#coding:utf-8

import threading
import time

loops = (4,2)

class MyThread(threading.Thread):
    def __init__(self,func,args,name=''):
        #threading.Thread.__init__(self)  
        super(MyThread,self).__init__()  #python 2
        #super().__init__() python 3
        self.name = name
        self.func = func
        self.args = args

    def run(self):
        apply(self.func,self.args)

def loop(nloop,nsec): #定义函数，打印运行的起止时间
    print "start loop %d,at %s" % (nloop,time.ctime())
    time.sleep(nsec)
    print "loop %d done at %s " % (nloop,time.ctime())

def main():
     print 'starting at: %s' % time.ctime()
     threads = []

     for i in range(2):
         t = MyThread(loop, (i, loops[i]),loop.__name__)
         threads.append(t) #创建两个线程，放入列表

     for i in range(2):
         threads[i].start()

     for i in range(2):
         threads[i].join()
         print 'all Done at %s' % time.ctime()

if __name__ == '__main__':
    main()

执行结果：

starting at: Wed Aug 21 10:50:48 2019
start loop 0,at Wed Aug 21 10:50:48 2019
start loop 1,at Wed Aug 21 10:50:48 2019
loop 1 done at Wed Aug 21 10:50:50 2019
loop 0 done at Wed Aug 21 10:50:52 2019
all Done at Wed Aug 21 10:50:52 2019
all Done at Wed Aug 21 10:50:52 2019
 

练习题

1、基于多线程实现并发的套接字通信

注意：只能在python3 中实现

客户端

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import socket

ip_port = ('127.0.0.1', 9999)
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(ip_port)

while True:
    cmd = input(">>>").strip()
    if not cmd:
        continue
    client.send(cmd.encode('utf-8'))
    data = client.recv(1024)
    print(data.decode('utf-8'))
client.close()

服务端

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import multiprocessing
import threading
import socket

ip_port = ('127.0.0.1', 9999)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(ip_port)
s.listen(5)


def action(conn):
    while True:
        data = conn.recv(1024)
        print(data)
        conn.send(data.upper())


if __name__ == '__main__':
    while True:
        conn, addr = s.accept()

        p = threading.Thread(target=action, args=(conn,))
        p.start()

执行结果：

客户端：
C:Python36python3.exe G:/PycharmProject/test/test/cli.py
>>>dir
DIR
>>>ipconfig
IPCONFIG
>>>hello
HELLO
>>>

服务端：
C:Python36python3.exe G:/PycharmProject/test/test/ser.py
b'dir'
b'ipconfig'
b'hello'


2、编写一个简单的文本处理工具，具备三个任务，一个接收用户输入，一个将用户输入的内容格式化成大写，一个将格式化后的结果存入文件

实现的功能是追加内容到某一个文件，文件不存在会先创建再追加

#!/usr/bin/env python
#coding:utf-8

# 练习二：三个任务，一个接收用户输入，一个将用户输入的内容格式
# 化成大写，一个将格式化后的结果存入文件

from threading import Thread

msg_l = []
format_l = []


def talk():
    while True:
        msg = input(">>>").strip()
        if not msg:
            break
        msg_l.append(msg)


def format_msg():
    while True:
        if msg_l:
            res = msg_l.pop()
            format_l.append(res.upper())


def save():
    while True:
        if format_l:
            with open('db.txt', 'a', encoding='utf-8') as f:
                res = format_l.pop()
                f.write('%s
' % res)


if __name__ == '__main__':
    t1 = Thread(target=talk)
    t2 = Thread(target=format_msg)
    t3 = Thread(target=save)

    t1.start()
    t2.start()
    t3.start()

多线程与多进程的区别

下面说一下多线程和多进程的区别

谁的开启速度快？（说明开进程的开销远远大于开线程，因为进程要申请内存空间）

 

同一进程内的线程都是平级的，不存在子线程之说，只有进程才存在主进程和子进程之说

开启进程，操作系统要申请内存空间，让好拷贝父进程地址空间到子进程，开销远大于线程

在主进程下开启线程

例子：

#!/usr/bin/env python
#coding:utf-8

import time
import random
from multiprocessing import Process



def study(name):
    print("%s is learning" % name)
    time.sleep(random.randint(1, 3))
    print("%s is playing" % name)


if __name__ == '__main__':
    t = Process(target=study, args=('james',))
    t.start()
    print("主进程开始运行....")

执行结果如下，几乎是t.start ()的同时就将线程开启了，然后先打印出了主进程程开始运行....，证明线程的创建开销极小

主进程开始运行....
james is learning
james is playing

在主进程下开启子进程

例子：

#!/usr/bin/env python
#coding:utf-8
import time
import random
from threading import Thread


def study(name):
    print("%s is learning" % name)
    time.sleep(random.randint(1, 3))
    print("%s is playing" % name)


if __name__ == '__main__':
    t = Thread(target=study, args=('james',))
    t.start()
    print("主线程开始运行....")

执行结果如下，p.start ()将开启进程的信号发给操作系统后，操作系统要申请内存空间，让好拷贝父进程地址空间到子进程，开销远大于线程

james is learning
主线程开始运行....
james is playing

开一下PID

在主进程下开启多个线程，每个线程都跟主进程的pid一样（线程共享主进程的Pid）

#!/usr/bin/env python
#coding:utf-8

from threading import Thread
import os
 
def work():
    print('hello',os.getpid())
 
if __name__ == '__main__':
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print('主线程/主进程pid',os.getpid())

执行结果：

hello 7939
hello 7939
主线程/主进程 7939

开多个进程，每个进程都有不同的pid

#!/usr/bin/env python
#coding:utf-8
from multiprocessing import Process
import os
 
def work():
    print('hello',os.getpid())
 
if __name__ == '__main__':
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print('主线程/主进程',os.getpid())

执行结果：

主线程/主进程 7951
hello 7952
hello 7953

同一进程内的线程共享该进程的数据

进程之间地址空间是隔离的

#!/usr/bin/env python
#coding:utf-8
from multiprocessing import Process
import os
 
def work():
    global n
    n=0
 
if __name__ == '__main__':
    n=100
    p=Process(target=work)
    p.start()
    p.join()
    print('主',n)

执行结果如下，毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100

主 100

同一进程内开启的多个线程是共享该进程地址空间的

#!/usr/bin/env python
#coding:utf-8
from threading import Thread
import os
 
def work():
    global n
    n=0
 
if __name__ == '__main__':
    n=100
    t=Thread(target=work)
    t.start()
    t.join()
    print('主',n)

执行结果如下， 查看结果为0,因为同一进程内的线程之间共享进程内的数据

主 0

Thread对象的其他属性或方法

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

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

验证

 代码1:



#!/usr/bin/env python
# coding:utf-8



from threading import Thread
from threading import current_thread
import time
 
def task():
    print("%s is running"% current_thread().getName())
    time.sleep(1)
    print("%s is done" % current_thread().getName())
 
if __name__ =='__main__':
    #没有子线程这个概念，只是为了理解方便
    t = Thread(target=task,name='子线程1')
    t.start()
    t.setName('儿子线程1')
    print("主线程 %s" % current_thread().getName())





执行结果：





主线程 MainThread
子线程1 is running
儿子线程1 is done





例子2：





#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import threading



def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())


if __name__ == '__main__':
    # 在主进程下开启线程
    t = Thread(target=work)
    t.start()

    print(threading.current_thread().getName())
    print(threading.current_thread())  # 主线程
    print(threading.enumerate())  # 连同主线程在内有两个运行的线程
    print(threading.active_count())
    print('主线程/主进程')

    '''
    打印结果:
    MainThread
    <_MainThread(MainThread, started 140735268892672)>
    [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
    主线程/主进程
    Thread-1
    '''






join方法：主线程等待子线程结束



例子：




#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time


def sayhi(name):
    time.sleep(2)
    print('%s say hello' % name)


if __name__ == '__main__':
    t = Thread(target=sayhi, args=('james',))
    t.start()
    t.join()
    print('主线程')
    print(t.is_alive())
    '''
    james say hello
    主线程
    False
    '''






python的执行顺序



 例子：




#!/usr/bin/env python
# coding:utf-8
import threading
import time

class MyThread(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "I'm "+self.name + "-" + str(i)
            print(msg)
def test():
    for i in range(5):
        t = MyThread()
        t.start()
if __name__ == '__main__':
    test()





执行结果：




I'm Thread-4-0
I'm Thread-2-0
I'm Thread-1-0
I'm Thread-3-0
I'm Thread-5-0
I'm Thread-4-1
I'm Thread-2-1
I'm Thread-1-1
I'm Thread-3-1
I'm Thread-5-1
I'm Thread-2-2
I'm Thread-4-2
I'm Thread-3-2
I'm Thread-1-2
I'm Thread-5-2





说明：




从代码和执行结果我们可以看出，多线程程序的执行顺序是不确定的。当执行到sleep语句时，线程将被阻塞（Blocked），到sleep结束后，线程进入就绪（Runnable）状态，
等待调度。而线程调度将自行选择一个线程执行。上面的代码中只能保证每个线程都运行完整个run函数，但是线程的启动顺序、run函数中每次循环的执行顺序都不能确定。






 




多线程-共享全局变量




 


 


例子：






#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time

g_num = 100

def work1():
    global g_num
    for i in range(3):
        g_num += 1

    print("----in work1, g_num is %d---"%g_num)


def work2():
    global g_num
    print("----in work2, g_num is %d---"%g_num)


print("---线程创建之前g_num is %d---"%g_num)

t1 = Thread(target=work1)
t1.start()

#延时一会，保证t1线程中的事情做完
time.sleep(1)

t2 = Thread(target=work2)
t2.start()

"""
---线程创建之前g_num is 100---
----in work1, g_num is 103---
----in work2, g_num is 103---
"""







 




列表当做实参传递到线程中




 


例子：


 


 






#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time

def work1(nums):
    nums.append(44)
    print("----in work1---",nums)


def work2(nums):
    #延时一会，保证t1线程中的事情做完
    time.sleep(1)
    print("----in work2---",nums)

g_nums = [11,22,33]

t1 = Thread(target=work1, args=(g_nums,))
t1.start()

t2 = Thread(target=work2, args=(g_nums,))
t2.start()

"""
----in work1--- [11, 22, 33, 44]
----in work2--- [11, 22, 33, 44]
"""







总结：






    在一个进程内的所有线程共享全局变量，能够在不适用其他方式的前提下完成多线程之间的数据共享（这点要比多进程要好）
    缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱（即线程非安全）






守护线程



无论是进程还是线程，都遵循：守护xx会等待主xx运行完毕后被销毁。


需要强调的是：运行完毕并非终止运行





#1.对主进程来说，运行完毕指的是主进程代码运行完毕
#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

详细介绍




#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，
　　才会结束，
#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，
　　而进程必须保证非守护线程都运行完毕后才能结束。







我们先来看一个例子





#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import os,time,random

def task():
    # t=Thread(target=time.sleep,args=(3,))
    # t.start()
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())

if __name__ == '__main__':
    t=Thread(target=task)
    t.daemon = True
    t.start()
    print('主')


"""
21048 is running
主
"""





原因是：




在执行到守护子线程t，由于主线程子线程通用一块内存，所以不存在不同进程创建各自空间，所以就先输出子进程的执行任务代码，
所以输出print(‘%s is running’ %os.getpid())，由于time.sleep(2)，所以就会执行主线程“main”,然后主线程执行完毕，那么即使2秒过后，
由于主线程执行完毕，那么子守护线程也就退出了，所以 print(‘%s is done’ %os.getpid())就不会执行了





守护子线程非守护子进程并存


我们解析来看一个守护子线程非守护子进程并存的例子








#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")

if __name__ == '__main__':
    t1=Thread(target=foo)
    t2 = Thread(target=bar)

    t1.daemon=True

    t2.start()
    t1.start()
    print("main-------")

"""
456
123
main-------
end123
end456
"""




原因是：




t1是守护子线程，t2非守护子线程，跟主线程使用一块内存，所以会输出t1,t1子线程的任务代码，所以执行456，123由于t1,t2都有睡眠时间，所以执行主线程代码，
然后对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕，所以会执行t1,t2睡眠后的任务代码，然后程序退出。
我们会问为什么t1守护子线程，也会执行sleep后的代码，不是说主线程代码执行完毕，守护线程就被干掉了吗？这里要注意是对主线程来说，
运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕，当时t2还没执行完毕










GIL全局解释器锁



一，介绍



定义：





In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple
native threads from executing Python bytecodes at once. This lock is necessary mainly
because CPython’s memory management is not thread-safe. (However, since the GIL
exists, other features have grown to depend on the guarantees that it enforces.)





结论：在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势






 首先需要明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。就好比C++是一套语言（语法）标准，
但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++，Visual C++等。Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。
像其中的JPython就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷。
 
所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL





二，GIL介绍


　　


GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。


可以肯定的一点是：保护不同的数据的安全，就应该加不同的锁。



要想了解GIL，首先确定一点：每次执行python程序，都会产生一个独立的进程。例如python test.py，python aaa.py，python bbb.py会产生3个不同的python进程



验证python test.py 只会产生一个进程





'''
#验证python test.py只会产生一个进程
#test.py内容
import os,time
print(os.getpid())
time.sleep(1000)
'''
python3 test.py
#在windows下
tasklist |findstr python
#在linux下
ps aux |grep python





在一个python的进程内，不仅有test.py的主线程或者由该主线程开启的其他线程，还有解释器开启的垃圾回收等解释器级别的线程，总之，所有线程都运行在这一个进程内，毫无疑问





  1 所有数据都是共享的，这其中，代码作为一种数据也是被所有线程共享的（test.py的所有代码以及Cpython解释器的所有代码）
 
 例如：test.py定义一个函数work（代码内容如下图），在进程内所有线程都能访问到work的代码，于是我们可以开启三个线程然后target都指向该代码，能访问到意味着就是可以执行。
 
    2 所有线程的任务，都需要将任务的代码当做参数传给解释器的代码去执行，即所有的线程要想运行自己的任务，首先需要解决的是能够访问到解释器的代码。







综上：





如果多个线程的target=work，那么执行流程是：多个线程先访问到解释器的代码，即拿到执行权限，然后将target的代码交给解释器的代码去执行






解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据100，可能线程1执行x=100的同时，而垃圾回收执行的是回收100的操作，解决这种问题没有什么高明的方法，就是加锁处理，如下图的GIL，保证python解释器同一时间只能执行一个任务的代码







三，GIL与Lock


　　机智的同学可能会问到这个问题：Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock?


　　首先，我们需要达成共识：锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据


　　然后，我们可以得出结论：保护不同的数据就应该加不同的锁。


　　最后，问题就很明朗了，GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock


　　GIL保护的是解释器级别的数据，保护用户自己的数据则需要自己加锁处理，如下图：










分析：




  1、100个线程去抢GIL锁，即抢执行权限
 
    2、肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()
 
    3、极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还未被线程1释放，于是阻塞，被迫交出执行权限，即释放GIL
 
    4、直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程






代码示例：





#!/usr/bin/env python
# coding:utf-8

from threading import Thread
from threading import Lock
import time

n = 100


def task():
    global n
    mutex.acquire()
    temp = n
    time.sleep(0.1)
    n = temp - 1
    mutex.release()


if __name__ == '__main__':
    mutex = Lock()
    t_l = []
    for i in range(100):
        t = Thread(target=task)
        t_l.append(t)
        t.start()

    for t in t_l:
        t.join()
    print("主", n)








结果：肯定为0，由原来的并发执行变为串行，牺牲了执行效率保证了数据安全，不加锁则结果可能为99




主 0






四，GIL与多线程


　　有了GIL的存在，同一时刻同一进程中只有一个线程被执行


　　听到这里，有的同学立马质问：进程可以利用多核，但是开销大，而python的多线程开销小，但却无法利用多核优势，也就是说python没用了？


 所以说 要解决这个问题，我们需要在几个点上达成一致：





1. cpu到底是用来做计算的，还是用来做I/O的？
 
2. 多cpu，意味着可以有多个核并行完成计算，所以多核提升的是计算性能
 
3. 每个cpu一旦遇到I/O阻塞，仍然需要等待，所以多核对I/O操作没什么用处 





一个工人相当于cpu，此时计算相当于工人在干活，I/O阻塞相当于为工人干活提供所需原材料的过程，工人干活的过程中如果没有原材料了，则工人干活的过程需要停止，直到等待原材料的到来。


　　如果你的工厂干的大多数任务都要有准备原材料的过程（I/O密集型），那么你有再多的工人，意义也不大，还不如一个人，在等材料的过程中让工人去干别的活，反过来讲，如果你的工厂原材料都齐全，那当然是工人越多，效率越高


结论：





对计算来说，cpu越多越好，但是对于I/O来说，再多的cpu也没用
 
　　当然对运行一个程序来说，随着cpu的增多执行效率肯定会有所提高（不管提高幅度多大，总会有所提高），这是因为一个程序基本上不会是纯计算或者
纯I/O，所以我们只能相对的去看一个程序到底是计算密集型还是I/O密集型，从而进一步分析python的多线程到底有无用武之地







假设我们有四个任务需要处理，处理方式肯定是需要玩出并发的效果，解决方案可以是：





方案一：开启四个进程
 
方案二：一个进程下，开启四个线程





单核情况下，分析结果：





如果四个任务是计算密集型，没有多核来并行计算，方案一徒增了创建进程的开销。方案二胜
 
如果四个任务是I/O密集型，方案一创建进程的开销大，且进程的切换速度远不如线程，方案二胜





多核情况下，分析结果：





如果四个任务是计算密集型，多核意味着并行计算，在python中一个进程中同一时刻只有一个线程执行，并不算多核。方案一胜
 
如果四个任务是I/O密集型，再多的核也解决不了I/O问题，方案二胜





结论：





    现在的计算机基本上都是多核，python对于计算密集型的任务开多线程的效率并不能带来多大性能上的提升，甚至不如串行(没有大量切换)，
但是，对于IO密集型的任务效率还是有显著提升的。





五，多线程性能测试


 


如果并发的多个任务是计算密集型：多进程效率高



（1）计算密集型实例


　　


多线程：




#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time
def work():
    res=0
    for i in range(100000000):
        res*=i
if __name__ == '__main__':
    start=time.time()
    result=[]
    for i in range(4):#因为4核，所以开启4个进程
        p=Thread(target=work)
        result.append(p)
        p.start()
    for p in result:
        p.join()
    end=time.time()
    print(end-start)  #32.61332559585571





#多进程:




#!/usr/bin/env python
# coding:utf-8

from multiprocessing import Process
import time
def work():
    res=0
    for i in range(100000000):
        res*=i
if __name__ == '__main__':
    start=time.time()
    result=[]
    for i in range(4):#因为4核，所以开启4个进程
        p=Process(target=work)
        result.append(p)
        p.start()
    for p in result:
        p.join()
    end=time.time()
    print(end-start)  #18.899179697036743





分析：对于多进程情况，所开4个进程分别在4个cpu上同时进行执行，所花时间为开启进程时间和单个进程运行时间的总和，对于多线程情况，其都在竞争解释器权限，排队进行执行代码，所化时间为四个线程的总和。多线程花费时间未与多进程花费时间呈4倍关系是因为开启进程开销比开启线程开销大很多。



（2）IO密集型实例



如果并发的多个任务是I/O密集型：多线程效率高


  


多线程：




#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time
def work():
   time.sleep(2)
if __name__ == '__main__':
    start=time.time()
    result=[]
    for i in range(400):
        p=Thread(target=work)
        result.append(p)
        p.start()
    for p in result:
        p.join()
    end=time.time()
    print(end-start)  #2.066821813583374





多进程：





#!/usr/bin/env python
# coding:utf-8

from multiprocessing import Process
import time
def work():
    time.sleep(2)

if __name__ == '__main__':
    start = time.time()
    result = []
    for i in range(400):
        p = Process(target=work)
        result.append(p)
        p.start()
    for p in result:
        p.join()
    end = time.time()
    print(end - start)  #78.71693396568298




分析：对于多线程情况，线程遇到阻塞，便会切换到另个线程，所化总时间就为花时间最长的单个线程的时间，对于多进程情况，同时开启这么多进程会花费很多时间，其次单个进程执行遇到阻塞时，若没有其他任务也会继续等待阻塞结束。


 





应用：





多线程用于IO密集型，如socket 爬虫 ，web
 
多进程用于计算密集型，如金融分析






同步



1.多线程开发可能遇到的问题




 


 




假设两个线程t1和t2都要对num=0进行增1运算，t1和t2都各对num修改10次，num的最终的结果应该为20。


但是由于是多线程访问，有可能出现下面情况：


在num=0时，t1取得num=0。此时系统把t1调度为”sleeping”状态，把t2转换为”running”状态，t2也获得num=0。然后t2对得到的值进行加1并赋给num，使得num=1。然后系统又把t2调度为”sleeping”，把t1转为”running”。线程t1又把它之前得到的0加1后赋值给num。这样，明明t1和t2都完成了1次加1工作，但结果仍然是num=1。





#!/usr/bin/env python
# coding:utf-8

from threading import Thread
import time

g_num = 0

def test1():
    global g_num
    for i in range(1000000):
        g_num += 1

    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    for i in range(1000000):
        g_num += 1

    print("---test2---g_num=%d"%g_num)


p1 = Thread(target=test1)
p1.start()

#time.sleep(3) #取消屏蔽之后 再次运行程序，结果会不一样，，，为啥呢？

p2 = Thread(target=test2)
p2.start()

print("---g_num=%d---"%g_num)




运行结果(可能不一样，但是结果往往不是2000000)：







---g_num=66918---
---test2---g_num=1052929
---test1---g_num=1108692





取消屏蔽之后，再次运行结果如下：




---test1---g_num=1000000
---g_num=1012302---
---test2---g_num=2000000





问题产生的原因就是没有控制多个线程对同一资源的访问，对数据造成破坏，使得线程运行的结果不可预期。这种现象称为“线程不安全”。 






2.什么是同步 


同步就是协同步调， 按预定的先后次序进⾏运⾏。如进程、 线程同步， 可理解为进程或线程A和B⼀块配合， A执⾏到⼀定程度时要依靠B的某个结果， 


于是停下来， 示意B运⾏;B依⾔执⾏， 再将结果给A;A再继续操作 




解决问题的思路 



对于提出的那个计算错误的问题，可以通过线程同步来进行解决



思路如下:





1.系统调用t1，然后获取到num的值为0，此时上一把锁，即不允许其他现在操作num
2.对num的值进行+1解锁，此时num的值为1，其他的线程就可以使用num了，而且是num的值不是0而是1
3.同理其他线程在对num进行修改时，都要先上锁，处理完后再解锁，在上锁的整个过程中不允许其他线程访问，就保证了数据的正确性








互斥锁 



当多个线程几乎同时修改某一个共享数据的时候，需要进行同步控制


线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。


互斥锁为资源引入一个状态：锁定/非锁定。


某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。



threading模块中定义了Lock类， 可以⽅便的处理锁定： 





 


#创建锁
mutex = threading.Lock()
#锁定
mutex.acquire([blocking])
#释放
mutex.release()


 






其中，锁定方法acquire可以有一个blocking参数。




    如果设定blocking为True，则当前线程会堵塞，直到获取到这个锁为止（如果没有指定，那么默认为True）
    如果设定blocking为False，则当前线程不会堵塞






使用互斥锁实现上面的例子的代码如下：





#!/usr/bin/env python
#coding:utf-8

from threading import Thread, Lock
import time

g_num = 0

def test1():
    global g_num
    for i in range(1000000):
        #True表示堵塞 即如果这个锁在上锁之前已经被上锁了，那么这个线程会在这里一直等待到解锁为止
        #False表示非堵塞，即不管本次调用能够成功上锁，都不会卡在这,而是继续执行下面的代码
        mutexFlag = mutex.acquire(True)
        if mutexFlag:
            g_num += 1
            mutex.release()

    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    for i in range(1000000):
        mutexFlag = mutex.acquire(True) #True表示堵塞
        if mutexFlag:
            g_num += 1
            mutex.release()

    print("---test2---g_num=%d"%g_num)

#创建一个互斥锁
#这个所默认是未上锁的状态
mutex = Lock()

p1 = Thread(target=test1)
p1.start()


p2 = Thread(target=test2)
p2.start()

print("---g_num=%d---"%g_num)





 



执行结果：




---g_num=27630---
---test1---g_num=1971147
---test2---g_num=2000000





可以看到，加入互斥锁后，运行结果与预期相符。




上锁解锁过程


当一个线程调用锁的acquire()方法获得锁时，锁就进入“locked”状态。


每次只有一个线程可以获得锁。如果此时另一个线程试图获得这个锁，该线程就会变为“blocked”状态，称为“阻塞”，直到拥有锁的线程调用锁的release()方法释放锁之后，锁进入“unlocked”状态。


线程调度程序从处于同步阻塞状态的线程中选择一个来获得锁，并使得该线程进入运行（running）状态。




总结


锁的好处：




确保了某段关键代码只能由一个线程从头到尾完整地执行





锁的坏处：




    阻止了多线程并发执行，包含锁的某段代码实际上只能以单线程模式执行，效率就大大地下降了
    由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁时，可能会造成死锁





多线程-非共享数据




对于全局变量，在多线程中要格外小心，否则容易造成数据错乱的情况发生



1. 非全局变量是否要加锁呢？



例子：




#!/usr/bin/env python
# coding:utf-8

import threading
import time

class MyThread(threading.Thread):
    # 重写 构造方法
    def __init__(self,num,sleepTime):
        threading.Thread.__init__(self)
        self.num = num
        self.sleepTime = sleepTime

    def run(self):
        self.num += 1
        time.sleep(self.sleepTime)
        print('线程(%s),num=%d'%(self.name, self.num))

if __name__ == '__main__':
    mutex = threading.Lock()
    t1 = MyThread(100,5)
    t1.start()
    t2 = MyThread(200,1)
    t2.start()
    
"""
线程(Thread-2),num=201
线程(Thread-1),num=101
"""





例子2：




#!/usr/bin/env python
# coding:utf-8

import threading
from time import sleep


def test(sleepTime):
    num = 1
    sleep(sleepTime)
    num += 1
    print('---(%s)--num=%d' % (threading.current_thread(), num))


t1 = threading.Thread(target=test, args=(5,))
t2 = threading.Thread(target=test, args=(1,))

t1.start()
t2.start()

"""
---(<Thread(Thread-2, started 37484)>)--num=2
---(<Thread(Thread-1, started 21652)>)--num=2
"""







总结：






在多线程开发中，全局变量是多个线程都共享的数据，而局部变量等是各自线程的，是非共享的







 


 






死锁


所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程



在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。


尽管死锁很少发生，但一旦发生就会造成应用的停止响应。下面看一个死锁的例子


 例子1：




 


#!/usr/bin/env python
#coding:utf-8


import threading
import time

class MyThread1(threading.Thread):
    def run(self):
        if mutexA.acquire():
            print(self.name+'----do1---up----')
            time.sleep(1)

            if mutexB.acquire():
                print(self.name+'----do1---down----')
                mutexB.release()
            mutexA.release()

class MyThread2(threading.Thread):
    def run(self):
        if mutexB.acquire():
            print(self.name+'----do2---up----')
            time.sleep(1)
            if mutexA.acquire():
                print(self.name+'----do2---down----')
                mutexA.release()
            mutexB.release()

mutexA = threading.Lock()
mutexB = threading.Lock()

if __name__ == '__main__':
    t1 = MyThread1()
    t2 = MyThread2()
    t1.start()
    t2.start()


 





执行结果：




Thread-1----do1---up----
Thread-2----do2---up----




此时已经进入到了死锁状态，可以使用ctrl-z退出



2.说明


如图：






3. 避免死锁




    程序设计时要尽量避免（银行家算法）
    添加超时时间等







 




附录-银行家算法


[背景知识]


一个银行家如何将一定数目的资金安全地借给若干个客户，使这些客户既能借到钱完成要干的事，同时银行家又能收回全部资金而不至于破产，这就是银行家问题。这个问题同操作系统中资源分配问题十分相似：银行家就像一个操作系统，客户就像运行的进程，银行家的资金就是系统的资源。


[问题的描述]


一个银行家拥有一定数量的资金，有若干个客户要贷款。每个客户须在一开始就声明他所需贷款的总额。若该客户贷款总额不超过银行家的资金总数，银行家可以接收客户的要求。客户贷款是以每次一个资金单位（如1万RMB等）的方式进行的，客户在借满所需的全部单位款额之前可能会等待，但银行家须保证这种等待是有限的，可完成的。


例如：有三个客户C1，C2，C3，向银行家借款，该银行家的资金总额为10个资金单位，其中C1客户要借9各资金单位，C2客户要借3个资金单位，C3客户要借8个资金单位，总计20个资金单位。某一时刻的状态如图所示。





对于a图的状态，按照安全序列的要求，我们选的第一个客户应满足该客户所需的贷款小于等于银行家当前所剩余的钱款，可以看出只有C2客户能被满足：C2客户需1个资金单位，小银行家手中的2个资金单位，于是银行家把1个资金单位借给C2客户，使之完成工作并归还所借的3个资金单位的钱，进入b图。同理，银行家把4个资金单位借给C3客户，使其完成工作，在c图中，只剩一个客户C1，它需7个资金单位，这时银行家有8个资金单位，所以C1也能顺利借到钱并完成工作。最后（见图d）银行家收回全部10个资金单位，保证不赔本。那麽客户序列{C1，C2，C3}就是个安全序列，按照这个序列贷款，银行家才是安全的。否则的话，若在图b状态时，银行家把手中的4个资金单位借给了C1，则出现不安全状态：这时C1，C3均不能完成工作，而银行家手中又没有钱了，系统陷入僵持局面，银行家也不能收回投资。


综上所述，银行家算法是从当前状态出发，逐个按安全序列检查各客户谁能完成其工作，然后假定其完成工作且归还全部贷款，再进而检查下一个能完成工作的客户，......。如果所有客户都能完成工作，则找到一个安全序列，银行家才是安全的。




 同步应用




多个线程有序执行



例子：




from threading import Thread,Lock
from time import sleep

class Task1(Thread):
    def run(self):
        while True:
            if lock1.acquire():
                print("------Task 1 -----")
                sleep(0.5)
                lock2.release()

class Task2(Thread):
    def run(self):
        while True:
            if lock2.acquire():
                print("------Task 2 -----")
                sleep(0.5)
                lock3.release()

class Task3(Thread):
    def run(self):
        while True:
            if lock3.acquire():
                print("------Task 3 -----")
                sleep(0.5)
                lock1.release()

#使用Lock创建出的锁默认没有“锁上”
lock1 = Lock()
#创建另外一把锁，并且“锁上”
lock2 = Lock()
lock2.acquire()
#创建另外一把锁，并且“锁上”
lock3 = Lock()
lock3.acquire()

t1 = Task1()
t2 = Task2()
t3 = Task3()

t1.start()
t2.start()
t3.start()





运行结果:




------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
------Task 1 -----
------Task 2 -----
------Task 3 -----
...省略...





总结





可以使用互斥锁完成多个任务，有序的进程工作，这就是线程的同步






递归锁RLock



死锁的解决方法是是使用递归锁；递归锁是在Python中为了支持在同一线程中多次请求同一资源，提供了可重入锁RLock。


这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。RLock和Lock的区别是：递归锁可以连续acquire多次，而互斥锁只能acquire一次。





#!/usr/bin/env python
# coding:utf-8

from threading import Thread, RLock
import time

mutexA = mutexB = RLock()


# 一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，
# 这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()

    def func1(self):
        mutexA.acquire()
        print('33[41m%s 拿到A锁33[0m' % self.name)

        mutexB.acquire()
        print('33[42m%s 拿到B锁33[0m' % self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('33[43m%s 拿到B锁33[0m' % self.name)
        time.sleep(2)

        mutexA.acquire()
        print('33[44m%s 拿到A锁33[0m' % self.name)
        mutexA.release()

        mutexB.release()


if __name__ == '__main__':
    for i in range(10):
        t = MyThread()
        t.start()





执行结果：




Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-1 拿到A锁
Thread-2 拿到A锁
Thread-2 拿到B锁
Thread-2 拿到B锁
Thread-2 拿到A锁
Thread-4 拿到A锁
Thread-4 拿到B锁
Thread-4 拿到B锁
Thread-4 拿到A锁
Thread-6 拿到A锁
Thread-6 拿到B锁
Thread-6 拿到B锁
Thread-6 拿到A锁
Thread-8 拿到A锁
Thread-8 拿到B锁
Thread-8 拿到B锁
Thread-8 拿到A锁
Thread-10 拿到A锁
Thread-10 拿到B锁
Thread-10 拿到B锁
Thread-10 拿到A锁
Thread-5 拿到A锁
Thread-5 拿到B锁
Thread-5 拿到B锁
Thread-5 拿到A锁
Thread-9 拿到A锁
Thread-9 拿到B锁
Thread-9 拿到B锁
Thread-9 拿到A锁
Thread-7 拿到A锁
Thread-7 拿到B锁
Thread-7 拿到B锁
Thread-7 拿到A锁
Thread-3 拿到A锁
Thread-3 拿到B锁
Thread-3 拿到B锁
Thread-3 拿到A锁







典型问题：科学家吃面，死锁问题





#!/usr/bin/env python
# coding:utf-8
import time
from threading import Thread,Lock
noodle_lock = Lock()
fork_lock = Lock()
def eat1(name):
    noodle_lock.acquire()
    print('%s 抢到了面条'%name)
    fork_lock.acquire()
    print('%s 抢到了叉子'%name)
    print('%s 吃面'%name)
    fork_lock.release()
    noodle_lock.release()

def eat2(name):
    fork_lock.acquire()
    print('%s 抢到了叉子' % name)
    time.sleep(1)
    noodle_lock.acquire()
    print('%s 抢到了面条' % name)
    print('%s 吃面' % name)
    noodle_lock.release()
    fork_lock.release()

for name in ['哪吒','egon','yuan']:
    t1 = Thread(target=eat1,args=(name,))
    t2 = Thread(target=eat2,args=(name,))
    t1.start()
    t2.start()





执行结果会卡住，这就是死锁





递归锁解决死锁问题







#!/usr/bin/env python
# coding:utf-8



import time
from threading import Thread,RLock
fork_lock = noodle_lock = RLock()
def eat1(name):
    noodle_lock.acquire()
    print('%s 抢到了面条'%name)
    fork_lock.acquire()
    print('%s 抢到了叉子'%name)
    print('%s 吃面'%name)
    fork_lock.release()
    noodle_lock.release()

def eat2(name):
    fork_lock.acquire()
    print('%s 抢到了叉子' % name)
    time.sleep(1)
    noodle_lock.acquire()
    print('%s 抢到了面条' % name)
    print('%s 吃面' % name)
    noodle_lock.release()
    fork_lock.release()

for name in ['哪吒','egon','yuan']:
    t1 = Thread(target=eat1,args=(name,))
    t2 = Thread(target=eat2,args=(name,))
    t1.start()
    t2.start()









ThreadLocal




 


在多线程环境下，每个线程都有自己的数据。一个线程使用自己的局部变量比使用全局变量好，因为局部变量只有线程自己能看见，不会影响其他线程，而全局变量的修改必须加锁。


 


1. 使用函数传参的方法




 


但是局部变量也有问题，就是在函数调用的时候，传递起来很麻烦：


 


 






def process_student(name):
    std = Student(name)
    # std是局部变量，但是每个函数都要用它，因此必须传进去：
    do_task_1(std)
    do_task_2(std)

def do_task_1(std):
    do_subtask_1(std)
    do_subtask_2(std)

def do_task_2(std):
    do_subtask_2(std)
    do_subtask_2(std)







 




每个函数一层一层调用都这么传参数那还得了？用全局变量？也不行，因为每个线程处理不同的Student对象，不能共享。


2. 使用全局字典的方法


如果用一个全局dict存放所有的Student对象，然后以thread自身作为key获得线程对应的Student对象如何？








global_dict = {}

def std_thread(name):
    std = Student(name)
    # 把std放到全局变量global_dict中：
    global_dict[threading.current_thread()] = std
    do_task_1()
    do_task_2()

def do_task_1():
    # 不传入std，而是根据当前线程查找：
    std = global_dict[threading.current_thread()]
    ...

def do_task_2():
    # 任何函数都可以查找出当前线程的std变量：
    std = global_dict[threading.current_thread()]
    ...





这种方式理论上是可行的，它最大的优点是消除了std对象在每层函数中的传递问题，但是，每个函数获取std的代码有点low。


有没有更简单的方式？



3. 使用ThreadLocal的方法


ThreadLocal应运而生，不用查找dict，ThreadLocal帮你自动做这件事




# !/usr/bin/env python
# coding:utf8

import threading

# 创建全局ThreadLocal对象:
local_school = threading.local()

def process_student():
    # 获取当前线程关联的student:
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))

def process_thread(name):
    # 绑定ThreadLocal的student:
    local_school.student = name
    process_student()

t1 = threading.Thread(target= process_thread, args=('dongGe',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('老王',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()

"""
Hello, dongGe (in Thread-A)
Hello, 老王 (in Thread-B)
"""







 




说明


全局变量local_school就是一个ThreadLocal对象，每个Thread对它都可以读写student属性，但互不影响。你可以把local_school看成全局变量，但每个属性如local_school.student都是线程的局部变量，可以任意读写而互不干扰，也不用管理锁的问题，ThreadLocal内部会处理。


可以理解为全局变量local_school是一个dict，不但可以用local_school.student，还可以绑定其他变量，如local_school.teacher等等。


ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，HTTP请求，用户身份信息等，这样一个线程的所有调用到的处理函数都可以非常方便地访问这些资源。



4. 小结


一个ThreadLocal变量虽然是全局变量，但每个线程都只能读写自己线程的独立副本，互不干扰。ThreadLocal解决了参数在一个线程中各个函数之间互相传递的问题








信号量（Semaphore）



信号量也是一把锁，互斥锁同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。公共厕所容纳的人数是一定的，这便是信号量的大小


信号量用来控制线程并发数的，BoundedSemaphore或Semaphore管理一个内置的计数 器，每当调用acquire()时-1，调用release()时+1。


  计数器不能小于0，当计数器为 0时，acquire()将阻塞线程至同步锁定状态，直到其他线程调用release()。(类似于停车位的概念)


  BoundedSemaphore与Semaphore的唯一区别在于前者将在调用release()时检查计数 器的值是否超过了计数器的初始值，如果超过了将抛出一个异常。






#!/usr/bin/env python
# coding:utf-8

from threading import Thread, Semaphore
import threading
import time


def func():
    sm.acquire()
    print('%s get sm' % threading.current_thread().getName())
    time.sleep(3)
    sm.release()


if __name__ == '__main__':
    sm = Semaphore(5)
    for i in range(23):
        t = Thread(target=func)
        t.start()







解析：





Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。






 


注意：


与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程


 


 





事件（event）


















同进程的一样，线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,


这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,


它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,


那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。


如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行






event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。





如图：










实例1：




#!/usr/bin/env python
# coding:utf-8

import threading,time
class Boss(threading.Thread):
    def run(self):
        print("BOSS：今晚大家都要加班到22:00。")
        event.isSet() or event.set()
        time.sleep(5)
        print("BOSS：<22:00>可以下班了。")
        event.isSet() or event.set()
class Worker(threading.Thread):
    def run(self):
        event.wait()
        print("Worker：哎……命苦啊！")
        time.sleep(0.25)
        event.clear()
        event.wait()
        print("Worker：OhYeah!")
if __name__=="__main__":
    event=threading.Event()
    threads=[]
    for i in range(5):
        threads.append(Worker())
    threads.append(Boss())
    for t in threads:
        t.start()
    for t in threads:
        t.join()
        
"""
BOSS：今晚大家都要加班到22:00。
Worker：哎……命苦啊！
Worker：哎……命苦啊！
Worker：哎……命苦啊！
Worker：哎……命苦啊！
Worker：哎……命苦啊！
BOSS：<22:00>可以下班了。
Worker：OhYeah!
Worker：OhYeah!
Worker：OhYeah!
Worker：OhYeah!
Worker：OhYeah!
"""






 


例如，有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作



示例：




#!/usr/bin/env python
# coding:utf-8

from threading import Event,current_thread,Thread
import time

def check():
    print('%s 正在检测' %current_thread().getName())
    time.sleep(3)
    event.set()

def conn():
    count=1
    while not event.is_set():
        if count > 3:
            raise TimeoutError('连接超时')
        print('%s 正在等待%s连接' % (current_thread().getName(),count))
        event.wait(timeout=1)    #设置阻塞时间
        count+=1
    print('%s 开始连接' % current_thread().getName())

if __name__ == '__main__':
    event = Event()
    t1=Thread(target=check)
    t2=Thread(target=conn)
    t3=Thread(target=conn)
    t4=Thread(target=conn)

    t1.start()
    t2.start()
    t3.start()
    t4.start()





 执行结果：




Thread-1 正在检测
Thread-2 正在等待1连接
Thread-3 正在等待1连接
Thread-4 正在等待1连接
Thread-4 正在等待2连接
Thread-3 正在等待2连接
Thread-2 正在等待2连接
Thread-4 正在等待3连接
Thread-3 正在等待3连接
Thread-2 正在等待3连接
Thread-4 开始连接
Thread-2 开始连接
Thread-3 开始连接







条件（Condition）



有一类线程需要满足条件之后才能够继续执行，Python提供了threading.Condition 对象用于条件变量线程的支持，它除了能提供RLock()或Lock()的方法外，还提供了 wait()、notify()、notifyAll()方法。


lock_con=threading.Condition([Lock/Rlock])： 锁是可选选项，不传入锁，对象自动创建一个RLock()。






wait()：条件不满足时调用，线程会释放锁并进入等待阻塞；
notify()：条件创造后调用，通知等待池激活一个线程；
notifyAll()：条件创造后调用，通知等待池激活所有线程。






 详细说明：




Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，
还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，
通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。







例子：




#!/usr/bin/env python
# coding:utf-8
import threading,time
from random import randint
class Producer(threading.Thread):
    def run(self):
        global L
        while True:
            val=randint(0,100)
            print('生产者',self.name,":Append"+str(val),L)
            if lock_con.acquire():
                L.append(val)
                lock_con.notify()
                lock_con.release()
            time.sleep(3)
class Consumer(threading.Thread):
    def run(self):
        global L
        while True:
                lock_con.acquire()
                if len(L)==0:
                    lock_con.wait()
                print('消费者',self.name,":Delete"+str(L[0]),L)
                del L[0]
                lock_con.release()
                time.sleep(0.25)

if __name__=="__main__":

    L=[]
    lock_con=threading.Condition()
    threads=[]
    for i in range(5):
        threads.append(Producer())
    threads.append(Consumer())
    for t in threads:
        t.start()
    for t in threads:
        t.join()








定时器



定时器，指定n秒后执行某个操作





#!/usr/bin/env python
# coding:utf-8

from threading import Timer


def hello():
    print("hello, world")


t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed











线程队列



queue队列 ：使用import queue，用法与进程Queue一样


当必须在多个线程之间安全地交换信息时，队列在线程编程中特别有用。



通用队列模块属性





 


1.Queue 模块函数 
queue(size) ：创建一个大小为 size 的 Queue 对象

2.Queue 对象函数
qsize() ：返回队列的大小（由于在返回的时候，队列可能会被其它线程修改，所以这个值是近似值）
empty() 如果队列为空返回 True，否则返回 False
full() 如果队列已满返回 True，否则返回 False
put(item,block=0) ：把 item 放到队列中，如果给了 block（不为 0），函数会一直阻塞到队列中有空间为止
get(block=0)： 从队列中取一个对象，如果给了 block（不为 0），函数会一直阻塞到队列中有对象为止


 






queue列队类的方法





 


#创建一个"队列"对象
import Queue
q = Queue.Queue(maxsize = 10)
Queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数maxsize来设定队列长度。
如果maxsize小于1就表示队列长度无限。

#将一个值放入队列中
q.put(10)
调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为1。
如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。

#将一个值从队列中取出
q.get()
调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。
如果队列为空且block为False，队列将引发Empty异常。

Python Queue模块有三种队列及构造函数:
1、Python Queue模块的FIFO队列先进先出。  class queue.Queue(maxsize)
2、LIFO类似于堆，即先进后出。             class queue.LifoQueue(maxsize)
3、还有一种是优先级队列级别越低越先出来。   class queue.PriorityQueue(maxsize)

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


 






 线程queue有三种不同的用法



1.class queue.Queue(maxsize=0) #先进先出



代码如下：





#!/usr/bin/env python
# coding:utf-8

import queue

q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
'''
结果(先进先出):
first
second
third
'''






2.class queue.LifoQueue(maxsize=0) #last in fisrt out


后进先出


 代码如下：





#!/usr/bin/env python
# coding:utf-8

import queue

q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
'''
结果(后进先出):
third
second
first
'''






3.class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列



优先级队列





#!/usr/bin/env python
# coding:utf-8
import queue

q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))

print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
(10, 'b')
(20, 'a')
(30, 'c')
'''








 



实例:





实现一个线程不断生成一个随机数到一个队列中(考虑使用Queue这个模块)

实现一个线程从上面的队列里面不断的取出奇数

实现另外一个线程从上面的队列里面不断取出偶数







 代码：


 




# !/usr/bin/env python
# coding:utf8
import random, threading, time
from Queue import Queue


# Producer thread
class Producer(threading.Thread):
    def __init__(self, t_name, queue):
        threading.Thread.__init__(self, name=t_name)
        self.data = queue

    def run(self):
        for i in range(10):  # 随机产生10个数字 ，可以修改为任意大小
            randomnum = random.randint(1, 99)
            print
            "%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), randomnum)
            self.data.put(randomnum)  # 将数据依次存入队列
            time.sleep(1)
        print
        "%s: %s finished!" % (time.ctime(), self.getName())


# Consumer thread
class Consumer_even(threading.Thread):
    def __init__(self, t_name, queue):
        threading.Thread.__init__(self, name=t_name)
        self.data = queue

    def run(self):
        while 1:
            try:
                val_even = self.data.get(1, 5)  # get(self, block=True, timeout=None) ,1就是阻塞等待,5是超时5秒
                if val_even % 2 == 0:
                    print
                    "%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val_even)
                    time.sleep(2)
                else:
                    self.data.put(val_even)
                    time.sleep(2)
            except:  # 等待输入，超过5秒 就报异常
                print
                "%s: %s finished!" % (time.ctime(), self.getName())
                break


class Consumer_odd(threading.Thread):
    def __init__(self, t_name, queue):
        threading.Thread.__init__(self, name=t_name)
        self.data = queue

    def run(self):
        while 1:
            try:
                val_odd = self.data.get(1, 5)
                if val_odd % 2 != 0:
                    print "%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val_odd)
                    time.sleep(2)
                else:
                    self.data.put(val_odd)
                    time.sleep(2)
            except:
                print "%s: %s finished!" % (time.ctime(), self.getName())
                break


# Main thread
def main():
    queue = Queue()
    producer = Producer('Pro.', queue)
    consumer_even = Consumer_even('Con_even.', queue)
    consumer_odd = Consumer_odd('Con_odd.', queue)
    producer.start()
    consumer_even.start()
    consumer_odd.start()
    producer.join()
    consumer_even.join()
    consumer_odd.join()
    print 'All threads terminate!'


if __name__ == '__main__':
    main()







 执行结果：




Wed Aug 21 17:33:58 2019: Con_odd. is consuming. 85 in the queue is consumed!


Wed Aug 21 17:34:00 2019: Con_odd. is consuming. 43 in the queue is consumed!





Wed Aug 21 17:34:04 2019: Con_odd. is consuming. 85 in the queue is consumed!





Wed Aug 21 17:34:08 2019: Con_odd. is consuming. 3 in the queue is consumed!


Wed Aug 21 17:34:10 2019: Con_odd. is consuming. 85 in the queue is consumed!
Wed Aug 21 17:34:12 2019: Con_odd. is consuming. 3 in the queue is consumed!

Wed Aug 21 17:34:19 2019: Con_odd. finished!
All threads terminate!







进程池与线程池


在刚开始学多进程或多线程时，我们迫不及待地基于多进程或多线程实现并发的套接字通信，然而这种实现方式的致命缺陷是：服务的开启的进程数或线程数都会随着并发的客户端数目地增多而增多，这会对服务端主机带来巨大的压力，甚至于不堪重负而瘫痪，于是我们必须对服务端开启的进程数或线程数加以控制，让机器在一个自己可以承受的范围内运行，这就是进程池或线程池的用途，例如进程池，就是用来存放进程的池子，本质还是基于多进程，只不过是对开启进程的数目加上了限制



Python标准模块--concurrent.futures



https://docs.python.org/dev/library/concurrent.futures.html



属性介绍：




concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.





2.基本方法：




#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数





 说明：




1.concurent.future模块是用来创建并行的任务，提供了更高级别的接口，
为了异步执行调用
2.concurent.future这个模块用起来非常方便，它的接口也封装的非常简单
3.concurent.future模块既可以实现进程池，也可以实现线程池
4.模块导入进程池和线程池
from  concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
还可以导入一个Executor,但是你别这样导，这个类是一个抽象类
抽象类的目的是规范他的子类必须有某种方法（并且抽象类的方法必须实现），但是抽象类不能被实例化
5.
  p = ProcessPoolExecutor(max_works)对于进程池如果不写max_works：默认的是cpu的数目,默认是4个
  p = ThreadPoolExecutor(max_works)对于线程池如果不写max_works：默认的是cpu的数目*5
6.如果是进程池，得到的结果如果是一个对象。我们得用一个.get()方法得到结果
  但是现在用了concurent.future模块，我们可以用obj.result方法
  p.submit(task,i)  #相当于apply_async异步方法
  p.shutdown()  #默认有个参数wite=True (相当于close和join)







 进程池和线程池



进程池：就是在一个进程内控制一定个数的线程
基于concurent.future模块的进程池和线程池 （他们的同步执行和异步执行是一样的）



基于concurent.future模块的进程池


1.同步执行





# !/usr/bin/env python
# coding:utf8

# 1.同步执行--------------
from  concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time,random
def task(n):
    print('[%s] is running'%os.getpid())
    time.sleep(random.randint(1,3))  #I/O密集型的，，一般用线程，用了进程耗时长
    return n**2
if __name__ == '__main__':
    start = time.time()
    p = ProcessPoolExecutor()
    for i in range(10): #现在是开了10个任务， 那么如果是上百个任务呢，就不能无线的开进程，那么就得考虑控制
        # 线程数了，那么就得考虑到池了
        obj  = p.submit(task,i).result()  #相当于apply同步方法
    p.shutdown()  #相当于close和join方法
    print('='*30)
    print(time.time() - start)  #17.36499309539795





执行结果：




[4636] is running
[16368] is running
[2300] is running
[1828] is running
[4636] is running
[16368] is running
[2300] is running
[1828] is running
[4636] is running
[16368] is running
==============================
17.9659998417





2.异步执行




# !/usr/bin/env python
# coding:utf8



#2.异步执行-----------
from  concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time,random
def task(n):
    print('[%s] is running'%os.getpid())
    time.sleep(random.randint(1,3))  #I/O密集型的，，一般用线程，用了进程耗时长
    return n**2
if __name__ == '__main__':
    start = time.time()
    p = ProcessPoolExecutor()
    l = []
    for i in range(10): #现在是开了10个任务， 那么如果是上百个任务呢，就不能无线的开进程，那么就得考虑控制
        # 线程数了，那么就得考虑到池了
        obj  = p.submit(task,i)  #相当于apply_async()异步方法
        l.append(obj)
    p.shutdown()  #相当于close和join方法
    print('='*30)
    print([obj.result() for obj in l])
    print(time.time() - start)  #5.362306594848633





执行结果：




[11896] is running
[15972] is running
[4232] is running
[9164] is running
[11896] is running
[15972] is running
[4232] is running
[9164] is running
[4232] is running
[9164] is running
==============================
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
7.23399996758






基于concurent.future模块的线程池





# !/usr/bin/env python
# coding:utf8

from  concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import currentThread
import os,time,random
def task(n):
    print('%s:%s is running'%(currentThread().getName(),os.getpid()))  #看到的pid都是一样的，因为线程是共享了一个进程
    time.sleep(random.randint(1,3))  #I/O密集型的，，一般用线程，用了进程耗时长
    return n**2
if __name__ == '__main__':
    start = time.time()
    p = ThreadPoolExecutor() #线程池 #如果不给定值，默认cup*5
    l = []
    for i in range(10):  #10个任务 # 线程池效率高了
        obj  = p.submit(task,i)  #相当于apply_async异步方法
        l.append(obj)
    p.shutdown()  #默认有个参数wite=True (相当于close和join)
    print('='*30)
    print([obj.result() for obj in l])
    print(time.time() - start)  #3.001171827316284





执行结果：




ThreadPoolExecutor-0_0:4636 is running
ThreadPoolExecutor-0_1:4636 is running
ThreadPoolExecutor-0_2:4636 is running
ThreadPoolExecutor-0_3:4636 is running
ThreadPoolExecutor-0_4:4636 is running
ThreadPoolExecutor-0_5:4636 is running
ThreadPoolExecutor-0_6:4636 is running
ThreadPoolExecutor-0_7:4636 is running
ThreadPoolExecutor-0_8:4636 is running
ThreadPoolExecutor-0_9:4636 is running
==============================
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
3.03299999237





应用线程池（下载网页并解析）





# !/usr/bin/env python
# coding:utf8

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import requests
import time,os
def get_page(url):
    print('<%s> is getting [%s]'%(os.getpid(),url))
    response = requests.get(url)
    if response.status_code==200:  #200代表状态：下载成功了
        return {'url':url,'text':response.text}
def parse_page(res):
    res = res.result()
    print('<%s> is getting [%s]'%(os.getpid(),res['url']))
    with open('db.txt','a') as f:
        parse_res = 'url:%s size:%s
'%(res['url'],len(res['text']))
        f.write(parse_res)
if __name__ == '__main__':
    # p = ThreadPoolExecutor()
    p = ProcessPoolExecutor()
    l = [
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
    ]
    for url in l:
        res = p.submit(get_page,url).add_done_callback(parse_page) #这里的回调函数拿到的是一个对象。得
        #  先把返回的res得到一个结果。即在前面加上一个res.result() #谁好了谁去掉回调函数
                                # 回调函数也是一种编程思想。不仅开线程池用，开线程池也用
    p.shutdown()  #相当于进程池里的close和join
    print('主',os.getpid())





执行结果：




<17256> is getting [http://www.baidu.com]
<4160> is getting [http://www.baidu.com]
<17164> is getting [http://www.baidu.com]
<16540> is getting [http://www.baidu.com]
<15532> is getting [http://www.baidu.com]
<15532> is getting [http://www.baidu.com]
<15532> is getting [http://www.baidu.com]
<15532> is getting [http://www.baidu.com]
主 15532







 







map的用法



map函数举例：




# map函数举例
obj= map(lambda x:x**2 ,range(10))
print(list(obj))

#运行结果[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]





可以和上面的开进程池/线程池的对比着看，就能发现map函数的强大了





# !/usr/bin/env python
# coding:utf8

# 我们的那个p.submit(task,i)和map函数的原理类似。我们就
# 可以用map函数去代替。更减缩了代码
from  concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time,random
def task(n):
    print('[%s] is running'%os.getpid())
    time.sleep(random.randint(1,3))  #I/O密集型的，，一般用线程，用了进程耗时长
    return n**2
if __name__ == '__main__':
    p = ProcessPoolExecutor()
    obj = p.map(task,range(10))
    p.shutdown()  #相当于close和join方法
    print('='*30)
    print(obj)  #返回的是一个迭代器
    print(list(obj))
    
"""
[1388] is running
[3388] is running
[16124] is running
[4688] is running
[3388] is running
[4688] is running
[1388] is running
[16124] is running
[1388] is running
[4688] is running
==============================
<generator object result_iterator at 0x000000000391BB40>
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
"""





例子2：






# !/usr/bin/env python
# coding:utf8

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import random,os,time

def task(name):
    print('name:%s pid:%s' %(name,os.getpid()))
    time.sleep(random.randint(1,3))


if __name__ == '__main__':
    p = ProcessPoolExecutor(2) # 进程池
    # t = ThreadPoolExecutor(2)# 线程池
    # for i in range(10):
    #     p.submit(task,'egon%s'%i)
    p.map(task,range(1,12)) #map取代了for+submit
    p.shutdown()
    print('主')








回调函数：



可以为进程池或线程池内的每个进程或线程绑定一个函数，该函数在进程或线程的任务执行完毕后自动触发，并接收任务的返回值当作参数，该函数称为回调函数


提交任务方式有两种：同步调用与异步调用


同步调用:提交完任务后，就在原地等待任务执行完毕，拿到结果，再执行下一行代码,导致程序是串行执行




# !/usr/bin/env python
# coding:utf8

from concurrent.futures import ThreadPoolExecutor
import time
import random

def la(name):
    print('%s is laing' %name)
    time.sleep(random.randint(3,5))
    res=random.randint(7,13)*'#'
    return {'name':name,'res':res}

def weigh(shit):
    name=shit['name']
    size=len(shit['res'])
    print('%s 拉了 《%s》kg' %(name,size))

if __name__ == '__main__':
    pool=ThreadPoolExecutor(13)

    shit1=pool.submit(la,'alex').result()
    weigh(shit1)
    shit2=pool.submit(la,'wupeiqi').result()
    weigh(shit2)
    shit3=pool.submit(la,'yuanhao').result()
    weigh(shit3)





执行结果：




alex is laing
alex 拉了 《11》kg
wupeiqi is laing
wupeiqi 拉了 《9》kg
yuanhao is laing
yuanhao 拉了 《8》kg





异步调用：提交完任务后，不地等待任务执行完毕





# !/usr/bin/env python
# coding:utf8

from concurrent.futures import ThreadPoolExecutor
import time
import random

def la(name):
    print('%s is laing' %name)
    time.sleep(random.randint(3,5))
    res=random.randint(7,13)*'#'
    return {'name':name,'res':res}


def weigh(shit):
    shit=shit.result()
    name=shit['name']
    size=len(shit['res'])
    print('%s 拉了 《%s》kg' %(name,size))


if __name__ == '__main__':
    pool=ThreadPoolExecutor(13)

    pool.submit(la,'alex').add_done_callback(weigh)

    pool.submit(la,'wupeiqi').add_done_callback(weigh)

    pool.submit(la,'yuanhao').add_done_callback(weigh)


"""
alex is laing
wupeiqi is laing
yuanhao is laing
yuanhao 拉了 《10》kg
wupeiqi 拉了 《12》kg
alex 拉了 《12》kg

"""





 进程池与线程池练习





# !/usr/bin/env python
# coding:utf8

from concurrent.futures import ThreadPoolExecutor
import time
import requests

def get(url):
    print('url:%s'%url)
    res = requests.get(url)
    time.sleep(3)
    return {'url':url,'content':res}


def pares(res):
    res = res.result()
    print('%s pares res is %s'%(res['url'],res['content']))

if __name__ == '__main__':
    urls = [
        'http://www.cnblogs.com/linhaifeng',
        'https://www.python.org',
        'https://www.openstack.org',
        'http://www.baidu.com'
    ]
    pool = ThreadPoolExecutor(2)
    for url in urls:
        pool.submit(get,url).add_done_callback(pares)
        
"""
url:http://www.cnblogs.com/linhaifeng
url:https://www.python.org
https://www.python.org pares res is <Response [200]>
url:https://www.openstack.org
http://www.cnblogs.com/linhaifeng pares res is <Response [200]>
url:http://www.baidu.com
http://www.baidu.com pares res is <Response [200]>
https://www.openstack.org pares res is <Response [200]>
"""











自定义线程池



low版线程池



 


设计思路：运用队列queue，

将线程类名放入队列中，执行一个就拿一个出来


注意：队列模块在python2 名为Queue,在python 3名为queue






# !/usr/bin/env python
# coding:utf8

import queue
import threading


class ThreadPool(object):

    def __init__(self, max_num=20):
        self.queue = queue.Queue(max_num)  #创建队列，最大数为20
        for i in range(max_num):
            self.queue.put(threading.Thread) #将类名放入队列中

    def get_thread(self):
        return self.queue.get() #从队列中取出类名

    def add_thread(self):
        self.queue.put(threading.Thread) #进类名放入队列中

def func(arg, p):  #定义一个函数
    print(arg)
    import time
    time.sleep(2)
    p.add_thread()


pool = ThreadPool(10) #创建对象，并执行该类的构造方法，即将线程的类名放入队列中

for i in range(30):
    thread = pool.get_thread() #调用该对象的get_thread方法，取出类名
    t = thread(target=func, args=(i, pool)) #创建对象，执行func，参数在args中
    t.start()





由于此方法要求使用者修改原函数，并在原函数里传参数，且调用方法也发生了改变，并且有空闲线程浪费资源，实际操作中并不方便，故设计了下一版线程池。



2.绝版线程池




设计思路：运用队列queue

a.队列里面放任务

b.线程一次次去取任务，线程一空闲就去取任务




# !/usr/bin/env python
# coding:utf8

import queue
import threading
import contextlib
import time

StopEvent = object()


class ThreadPool(object):

    def __init__(self, max_num, max_task_num = None):
        if max_task_num:
            self.q = queue.Queue(max_task_num)
        else:
            self.q = queue.Queue()
        self.max_num = max_num
        self.cancel = False
        self.terminal = False
        self.generate_list = []
        self.free_list = []

    def run(self, func, args, callback=None):
        """
        线程池执行一个任务
        :param func: 任务函数
        :param args: 任务函数所需参数
        :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
        :return: 如果线程池已经终止，则返回True否则None
        """
        if self.cancel:
            return
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            self.generate_thread()
        w = (func, args, callback,)
        self.q.put(w)

    def generate_thread(self):
        """
        创建一个线程
        """
        t = threading.Thread(target=self.call)
        t.start()

    def call(self):
        """
        循环去获取任务函数并执行任务函数
        """
        current_thread = threading.currentThread()
        self.generate_list.append(current_thread)

        event = self.q.get()
        while event != StopEvent:

            func, args, callback = event
            try:
                result = func(*args)
                success = True
            except Exception as e:
                success = False
                result = None

            if callback is not None:
                try:
                    callback(success, result)
                except Exception as e:
                    pass

            with self.worker_state(self.free_list, current_thread):
                if self.terminal:
                    event = StopEvent
                else:
                    event = self.q.get()
        else:

            self.generate_list.remove(current_thread)

    def close(self):
        """
        执行完所有的任务后，所有线程停止
        """
        self.cancel = True
        count = len(self.generate_list)
        while count:
            self.q.put(StopEvent)
            count -= 1

    def terminate(self):
        """
        无论是否还有任务，终止线程
        """
        self.terminal = True

        while self.generate_list:
            self.q.put(StopEvent)

        self.q.queue.clear()

    @contextlib.contextmanager
    def worker_state(self, state_list, worker_thread):
        """
        用于记录线程中正在等待的线程数
        """
        state_list.append(worker_thread)
        try:
            yield
        finally:
            state_list.remove(worker_thread)



# How to use

pool = ThreadPool(5)

def callback(status, result):
    # status, execute action status
    # result, execute action return value
    pass

def action(i):
    print(i)

for i in range(30):
    ret = pool.run(action, (i,), callback)

time.sleep(3)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
pool.close()
# pool.terminate()








练习：


创建多线程时间戳服务器 



编写 mttsServer.py 服务器脚本，主要要求如下：
1、服务器监听在 0.0.0.0 的 12345 端口上
2、收到客户端数据后，将其加上时间戳后回送给客户端
3、如果客户端发过来的字符全是空白字符，则终止与客户端的连接
4、要求能够同时处理多个客户端的请求
5、要求使用多线程的方式进行编写 




方案
多线程可以并行处理相互独立的子任务，大幅度地提升整个任务的效率。多数的线程服务器有同样的结构，


主线程是负责侦听请求的线程，主线程收到一个请求的时候，新的工作线程会被建立起来，处理客户端请求，客户端断开时，工作线程将终止。 




实现此案例需要按照如下步骤进行。 


步骤一：编写脚本 




 


#!/usr/bin/env python
#coding:utf-8

import socket
import os
import sys
import threading
import time

host = ''
port = 12345
addr = (host, port)


def handleChild(clientsock): #定义处理客户端连接请求的函数
    print 'New child', threading.currentThread().getName()
    print 'Got connection from:', clientsock.getpeername()

    while True:
        data = clientsock.recv(4096).strip()
        if not len(data):
            break
            
        clientsock.sendall('[%s] %s
' % (time.ctime(), data))
    clientsock.close()
    
    
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(addr)
s.listen(5)



while True:
    clientsock, clientaddr = s.accept()
    t = threading.Thread(target = handleChild, args = [clientsock])
    t.setDaemon(1)
    t.start()


 





步骤二：验证执行结果 




[root@py01 bin]# ./mttsServer.py #启动服务器




测试客户端：




 


[root@py01 bin]# telnet 127.0.0.1 12345 #可以打开多个终端，同时连接
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
hello world!
[Mon Jul 6 19:39:27 2015] hello world!
#此处输入回车，退出连接
Connection closed by foreign host


 






mtping1.py内容：




 


#!/usr/bin/env python
#coding:utf8

import os 
import threading

def ping(ip):
    result = os.system("ping -c2 %s &> /dev/null " % ip)
    if result == 0:
        print "%s:up" % ip 
    
    else:
        print "%s:down" % ip 
        
if __name__ == "__main__":
    ip_list = ("192.168.3.%s" % i for i in range(1,255))
    for ip in ip_list:
        ping(ip)


 





mtping2.py内容：




 


#!/usr/bin/env python
#coding:utf8

import os 
import threading

def ping(ip):
    result = os.system("ping -c2 %s &> /dev/null " % ip)
    if result == 0:
        print "%s:up" % ip 
    
    else:
        print "%s:down" % ip 
        
if __name__ == "__main__":
    ip_list = ("192.168.3.%s" % i for i in range(1,255))
    for ip in ip_list:
        t = threading.Thread(target=ping,args=[ip])
        t.start()


 





mt1.py内容：




 


#!/usr/bin/env python
#coding:utf8

import threading
import time 

def say_hi():
    time.sleep(5)
    print "hello"
    
if __name__ == "__main__":
    t = threading.Thread(target=say_hi)
    t.start()


 





mt2.py内容：





 


#!/usr/bin/env python
#coding:utf8

import threading
import time 

def say_hi():
    time.sleep(5)
    print "hello"
    
if __name__ == "__main__":
    t = threading.Thread(target=say_hi)
    t.setDaemon(1)
    t.start()


 






mttcpserv.py内容：




 


#!/usr/bin/env python
#coding:utf8

import socket
import time 
import threading

def handle_client(cli_sock):
    while True:
        data = cli_sock.recv(4096)
        if not data.strip():
            break
        
        cli_sock.send("[%s] %s" % (time.ctime(),data))
    cli_sock.close()
        
if __name__ =="__main__":
    host = ''
    port = 21345
    addr = (host,port)
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(addr)
    s.listen(1)
    
    while True:
        cli_sock,cli_addr = s.accept()
        t = threading.Thread(target=handle_client,args=[cli_sock])
        t.setDaemon(1)
        t.start()


 





mtping3.py内容：




 


#!/usr/bin/env python
#coding:utf8

import os 
import threading

class Ping(object):
    
    def __init__(self,ip):
        self.ip = ip 
        
        
    def __call__(self):
        result = os.system("ping -c2 %s &> /dev/null " % self.ip)
        if result == 0:
            print "%s:up" % self.ip 
        else:
            print "%s:down" % self.ip 
            
if __name__ == "__main__":
    ip_list = ("192.168.3.%s" % i for i in range(1,255))
    
    for ip in ip_list:
        t = threading.Thread(target=Ping(ip))
        t.start()


 





mtping4.py:


其中/root/work/newip.txt里面为IP，一行一个IP：





 


#!/usr/bin/env python
#coding:utf-8

import os
import threading

new_a = []



IP ="/root/work/newip.txt"
#with open("/root/work/newip.txt") as f1:
with open(IP) as f1:
    for i in f1.readlines():
        new_i = i.strip("
")
        new_a.append(new_i)

#print new_a

def ping(ip):
    result = os.system("ping -c1 %s &> /dev/null" % ip)
    if result == 0:
        print "%s:up" % ip
    else:
        print "%s:down" % ip


if __name__ == "__main__":
    for ip in new_a:
        #print ip
         t = threading.Thread(target=ping,args=[ip])
         t.start()


 





 mting5.py内容:


该文件实现了从指定IP里面将up状态的IP和down状态的IP分开


其中/root/work/newip.txt里面为IP，一行一个IP：


 




 


#!/usr/bin/env python
#coding:utf-8

import os
import threading

new_a = []
#exit_ip = []


IP ="/root/work/newip.txt"
#with open("/root/work/newip.txt") as f1:
with open(IP) as f1:
    for i in f1.readlines():
        new_i = i.strip("
")
        new_a.append(new_i)

#print new_a

def ping(ip):
    result = os.system("ping -c1 %s &> /dev/null" % ip)
    if result == 0:
        #print "%s:up" % ip
         with open('up.txt','a+') as f1:
                f1.write(ip + '
')
    else:
        #print "%s:down" % ip
         with open('down.txt','a+') as f2:
                f2.write(ip + '
')


if __name__ == "__main__":
    for ip in new_a:
        #print ip
         t = threading.Thread(target=ping,args=[ip])
         t.start()


 





参考：


https://www.cnblogs.com/wj-1314/p/9039970.html


 https://blog.csdn.net/u013210620/article/details/78710532


 https://www.cnblogs.com/seven-007/p/7662298.html


https://www.cnblogs.com/haiyan123/p/7461294.html


https://www.cnblogs.com/yjiu1990/p/9263267.html







        

	      

	    

	  

	  

	    
      
        【推广】
        免费学中医，健康全家人
      
	  

    

    

          原文地址：https://www.cnblogs.com/zhongguiyao/p/11049640.html