Python网络编程之线程与进程

What is a Thread？

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

在同一个进程内的线程的数据是可以进行互相访问的。

线程的切换使用过上下文来实现的，比如有一本书，有a和b这两个人(两个线程)看，a看完之后记录当前看到那一页哪一行，然后交给b看，b看完之后记录当前看到了那一页哪一行，此时a又要看了，那么a就通过上次记录的值(上下文)直接找到上次看到了哪里，然后继续往下看。

What is a Process？

一个进程至少要包含一个线程，每个进程在启动的时候就会自动的启动一个线程，进程里面的第一个线程就是主线程，每次在进程内创建的子线程都是由主线程进程创建和销毁，子线程也可以由主线程创建出来的线程创建和销毁线程。

进程是对各种资源管理的集合，比如要调用内存、CPU、网卡、声卡等，进程要操作上述的硬件之前都必须要创建一个线程，进程里面可以包含多个线程，QQ就是一个进程。

继续拿QQ来说，比如我现在打卡了QQ的聊天窗口、个人信息窗口、设置窗口等，那么每一个打开的窗口都是一个线程，他们都在执行不同的任务，比如聊天窗口这个线程可以和好友进行互动，聊天，视频等，个人信息窗口我可以查看、修改自己的资料。

为了进程安全起见，所以两个进程之间的数据是不能够互相访问的(默认情况下)，比如自己写了一个应用程序，然后让别人运行起来，那么我的这个程序就可以访问用户启动的其他应用，我可以通过我自己的程序去访问QQ，然后拿到一些聊天记录等比较隐秘的信息，那么这个时候就不安全了，所以说进程与进程之间的数据是不可以互相访问的，而且每一个进程的内存是独立的。

进程与线程的区别？

1. 线程是执行的指令集，进程是资源的集合

2. 线程的启动速度要比进程的启动速度要快

3. 两个线程的执行速度是一样的

4. 进程与线程的运行速度是没有可比性的

5. 线程共享创建它的进程的内存空间，进程的内存是独立的。

6. 两个线程共享的数据都是同一份数据，两个子进程的数据不是共享的，而且数据是独立的;

7. 同一个进程的线程之间可以直接交流，同一个主进程的多个子进程之间是不可以进行交流，如果两个进程之间需要通信，就必须要通过一个中间代理来实现;

8. 一个新的线程很容易被创建，一个新的进程创建需要对父进程进行一次克隆

9. 一个线程可以控制和操作同一个进程里的其他线程，线程与线程之间没有隶属关系，但是进程只能操作子进程

10. 改变主线程，有可能会影响到其他线程的行为，但是对于父进程的修改是不会影响子进程;

一个多并发的小脚本

1. import threading

2. import time

3. def Princ(String):

4.    print(‘task‘, String)

5.    time.sleep(5)

6. # target=目标函数， args=传入的参数

7. t1 = threading.Thread(target=Princ, args=(‘t1‘,))

8. t1.start()

9. t2 = threading.Thread(target=Princ, args=(‘t1‘,))

10. t2.start()

11. t3 = threading.Thread(target=Princ, args=(‘t1‘,))

12. t3.start()

参考文档

进程与线程的一个简单解释
http://www.ruanyifeng.com/blog/2013/04/processes_and_threads.html
Linux进程与线程的区别
https://my.oschina.net/cnyinlinux/blog/422207

线程

Thread module emulating a subset of Java’s threading model.

调用threading模块调用线程的两种方式

直接调用

1. import threading

2. import time

3. def Princ(String):

4.    print(‘task‘, String)

5.    time.sleep(5)

6. # target=目标函数， args=传入的参数

7. t1 = threading.Thread(target=Princ, args=(‘t1‘,))

8. t1.start()

9. t2 = threading.Thread(target=Princ, args=(‘t1‘,))

10. t2.start()

11. t3 = threading.Thread(target=Princ, args=(‘t1‘,))

12. t3.start()

通过类调用

1. import threading

2. import time

3. class MyThreading(threading.Thread):

4.    def __init__(self, conn):

5.        super(MyThreading, self).__init__()

6.        self.conn = conn

7.    def run(self):

8.        print(‘run task‘, self.conn)

9.        time.sleep(5)

10. t1 = MyThreading(‘t1‘)

11. t2 = MyThreading(‘t2‘)

12. t1.start()

13. t2.start()

多线程

多线程在Python内实则就是一个假象，为什么这么说呢，因为CPU的处理速度是很快的，所以我们看起来以一个线程在执行多个任务，每个任务的执行速度是非常之快的，利用上下文切换来快速的切换任务，以至于我们根本感觉不到。

但是频繁的使用上下文切换也是要耗费一定的资源，因为单线程在每次切换任务的时候需要保存当前任务的上下文。

什么时候用到多线程？

首先IO操作是不占用CPU的，只有计算的时候才会占用CPU(譬如1+1=2)，Python中的多线程不适合CPU密集型的任务，适合IO密集型的任务(sockt server)。

启动多个线程

主进程在启动之后会启动一个主线程，下面的脚本中让主线程启动了多个子线程，然而启动的子线程是独立的，所以主线程不会等待子线程执行完毕，而是主线程继续往下执行，并行执行。

1. for i in range(50):

2.    t = threading.Thread(target=Princ, args=(‘t-%s‘ % (i),))

3.    t.start()

join()

join()方法可以让程序等待每一个线程之后完成之后再往下执行，又成为串行执行。

1. import threading

2. import time

3. def Princ(String):

4.    print(‘task‘, String)

5.    time.sleep(1)

6. for i in range(50):

7.    t = threading.Thread(target=Princ, args=(‘t-%s‘ % (i),))

8.    t.start()

9.    # 当前线程执行完毕之后在执行后面的线程

10.    t.join()

让主线程阻塞，子现在并行执行

1. import threading

2. import time

3. def Princ(String):

4.    print(‘task‘, String)

5.    time.sleep(2)

6. # 执行子线程的时间

7. start_time = time.time()

8. # 存放线程的实例

9. t_objs = []

10. for i in range(50):

11.    t = threading.Thread(target=Princ, args=(‘t-%s‘ % (i),))

12.    t.start()

13.    # 为了不让后面的子线程阻塞，把当前的子线程放入到一个列表中

14.    t_objs.append(t)

15. # 循环所有子线程实例，等待所有子线程执行完毕

16. for t in t_objs:

17.    t.join()

18. # 当前时间减去开始时间就等于执行的过程中需要的时间

19. print(time.time() - start_time)

查看主线程与子线程

1. import threading

2. class MyThreading(threading.Thread):

3.    def __init__(self):

4.        super(MyThreading, self).__init__()

5.    def run(self):

6.        print(‘我是子线程： ‘, threading.current_thread())

7. t = MyThreading()

8. t.start()

9. print(‘我是主线程： ‘, threading.current_thread())

输出如下：

1. C:\Python\Python35\python.exe E:/MyCodeProjects/进程与线程/s3.py

2. 我是子线程：  <MyThreading(Thread-1, started 7724)>

3. 我是主线程：  <_MainThread(MainThread, started 3680)>

5. Process finished with exit code 0

查看当前进程的活动线程个数

1. import threading

2. class MyThreading(threading.Thread):

3.    def __init__(self):

4.        super(MyThreading, self).__init__()

5.    def run(self):

6.        print(‘www.anshengme.com‘)

7. t = MyThreading()

8. t.start()

9. print(‘线程个数： ‘, threading.active_count())

输出如下：

1. C:\Python\Python35\python.exe E:/MyCodeProjects/进程与线程/s3.py

2. www.anshengme.com

3. # 一个主线程和一个子线程

4. 线程个数：  2

6. Process finished with exit code 0

Event

Event是线程间通信最间的机制之一：一个线程发送一个event信号，其他的线程则等待这个信号。用于主线程控制其他线程的执行。 Events 管理一个flag，这个flag可以使用set
()设置成True或者使用clear()重置为False，wait()则用于阻塞，在flag为True之前。flag默认为False。

1. #!/use/bin/env python

2. # _*_ coding: utf-8- _*_

4. import threading

6. def runthreading(event):

7.    print("Start...")

8.    event.wait()

9.    print("End...")

10. event_obj = threading.Event()

11. for n in range(10):

12.    t = threading.Thread(target=runthreading, args=(event_obj,))

13.    t.start()

15. event_obj.clear()

16. inp = input("True/False?>> ")

17. if inp == "True":

18.    event_obj.set()

19. `

守护进程(守护线程)

一个主进程可以启动多个守护进程，但是主进程必须要一直运行，如果主进程挂掉了，那么守护进程也会随之挂掉

程序会等待主线程(进程)执行完毕，但是不会等待守护进程(线程)

1. import threading

2. import time

4. def Princ(String):

5.    print(‘task‘, String)

6.    time.sleep(2)

7. for i in range(50):

8.    t = threading.Thread(target=Princ, args=(‘t-%s‘ % (i),))

9.    t.setDaemon(True)  # 把当前线程设置为守护线程，要在start之前设置

10.    t.start()

场景预设： 比如现在有一个FTP服务，每一个用户连接上去的时候都会创建一个守护线程，现在已经有300个用户连接上去了，就是说已经创建了300个守护线程，但是突然之间FTP服务宕掉了，这个时候就不会等待守护线程执行完毕再退出，而是直接退出，如果是普通的线程，那么就会登台线程执行完毕再退出。

1. #!/use/bin/env python

2. # _*_ coding:utf-8 _*_

4. from multiprocessing import Process

5. import time

6. def runprocess(arg):

7.    print(arg)

8.    time.sleep(2)

11. p = Process(target=runprocess, args=(11,))

12. p.daemon=True

13. p.start()

15. print("end")

线程之间的数据交互与锁(互斥锁)

python2.x需要加锁，但是在python3.x上面就不需要了

1. # _*_ coding:utf-8 _*_

2. import threading

3. def Princ():

4.    # 获取锁

5.    lock.acquire()

6.    # 在函数内可以直接修改全局变量

7.    global number

8.    number += 1

9.    # 为了避免让程序出现串行，不能加sleep

10.    # time.sleep(1)

11.    # 释放锁

12.    lock.release()

13. # 锁

14. lock = threading.Lock()

15. # 主线程的number

16. number = 0

17. t_objs = []

18. for i in range(100):

19.    t = threading.Thread(target=Princ)

20.    t.start()

21.    t_objs.append(t)

22. for t in t_objs:

23.    t.join()

24. print(‘Number:‘, number)

递归锁(Lock/RLock)

1. import threading

2. def run1():

3.    print("grab the first part data")

4.    lock.acquire()

5.    global num

6.    num += 1

7.    lock.release()

8.    return num

9. def run2():

10.    print("grab the second part data")

11.    lock.acquire()

12.    global num2

13.    num2 += 1

14.    lock.release()

15.    return num2

16. def run3():

17.    lock.acquire()

18.    res = run1()

19.    print(‘--------between run1 and run2-----‘)

20.    res2 = run2()

21.    lock.release()

22.    print(res, res2)

23. t_objs = []

24. if __name__ == ‘__main__‘:

25.    num, num2 = 0, 0

26.    lock = threading.RLock()  # RLock()类似创建了一个字典，每次退出的时候找到字典的值进行退出

27.    # lock = threading.Lock()  # Lock()会阻塞在这儿

28.    for i in range(10):

29.        t = threading.Thread(target=run3)

30.        t.start()

31.        t_objs.append(t)

32. for t in t_objs:

33.    t.join()

34. print(num, num2)

信号量(Semaphore)

互斥锁同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据

1. import threading

2. import time

3. def run(n):

4.    semaphore.acquire()  # 获取信号，信号可以有多把锁

5.    time.sleep(1)  # 等待一秒钟

6.    print("run the thread: %s\n" % n)

7.    semaphore.release()  # 释放信号

8. t_objs = []

9. if __name__ == ‘__main__‘:

10.    semaphore = threading.BoundedSemaphore(5)  # 声明一个信号量，最多允许5个线程同时运行

11.    for i in range(20):  # 运行20个线程

12.        t = threading.Thread(target=run, args=(i,))  # 创建线程

13.        t.start()  # 启动线程

14.        t_objs.append(t)

15. for t in t_objs:

16.    t.join()

17. print(‘>>>>>>>>>>>>>‘)

以上代码中，类似与创建了一个队列，最多放5个任务，每执行完成一个任务就会往后面增加一个任务。

多进程

多进程的资源是独立的，不可以互相访问。

启动一个进程

1. from multiprocessing import Process

2. import time

3. def f(name):

4.    time.sleep(2)

5.    print(‘hello‘, name)

6. if __name__ == ‘__main__‘:

7.    # 创建一个进程

8.    p = Process(target=f, args=(‘bob‘,))

9.    # 启动

10.    p.start()

11.    # 等待进程执行完毕

12.    p.join(

在进程内启动一个线程

1. from multiprocessing import Process

2. import threading

3. def Thread(String):

4.    print(String)

5. def Proces(String):

6.    print(‘hello‘, String)

7.    t = threading.Thread(target=Thread, args=(‘Thread %s‘ % (String),))  # 创建一个线程

8.    t.start()  # 启动它

9. if __name__ == ‘__main__‘:

10.    p = Process(target=Proces, args=(‘World‘,))  # 创建一个进程

11.    p.start()  # 启动

12.    p.join()  # 等待进程执行完毕

启动一个多进程

1. from multiprocessing import Process

2. import time

3. def f(name):

4.    time.sleep(2)

5.    print(‘hello‘, name)

6. if __name__ == ‘__main__‘:

7.    for n in range(10):  # 创建一个进程

8.        p = Process(target=f, args=(‘bob %s‘ % (n),))

9.        # 启动

10.        p.start()

11.        # 等待进程执行完毕

获取启动进程的PID

1. # _*_ coding:utf-8 _*_

2. from multiprocessing import Process

3. import os

4. def info(String):

5.    print(String)

6.    print(‘module name:‘, __name__)

7.    print(‘父进程的PID:‘, os.getppid())

8.    print(‘子进程的PID:‘, os.getpid())

9.    print("\n")

10. def ChildProcess():

11.    info(‘\033[31;1mChildProcess\033[0m‘)

12. if __name__ == ‘__main__‘:

13.    info(‘\033[32;1mTheParentProcess\033[0m‘)

14.    p = Process(target=ChildProcess)

15.    p.start()

输出结果

1. C:\Python\Python35\python.exe E:/MyCodeProjects/多进程/s1.py

2. TheParentProcess

3. module name: __main__

4. # Pycharm的PID

5. 父进程的PID: 6888

6. # 启动的脚本PID

7. 子进程的PID: 4660

9. ChildProcess

10. module name: __mp_main__

11. # 脚本的PID

12. 父进程的PID: 4660

13. # 父进程启动的子进程PID

14. 子进程的PID: 8452

16. Process finished with exit code 0

进程间通信

默认情况下进程与进程之间是不可以互相通信的，若要实现互相通信则需要一个中间件，另个进程之间通过中间件来实现通信，下面是进程间通信的几种方式。

进程Queue

1. # _*_ coding:utf-8 _*_

2. from multiprocessing import Process, Queue

3. def ChildProcess(Q):

4.    Q.put([‘Hello‘, None, ‘World‘])  # 在Queue里面上传一个列表

5. if __name__ == ‘__main__‘:

6.    q = Queue()  # 创建一个Queue

7.    p = Process(target=ChildProcess, args=(q,))  # 创建一个子进程，并把Queue传给子进程,相当于克隆了一份Queue

8.    p.start()  # 启动子进程

9.    print(q.get())  # 获取q中的数据

10.    p.join()

管道(Pipes)

1. # _*_ coding:utf-8 _*_

2. from multiprocessing import Process, Pipe

3. def ChildProcess(conn):

4.    conn.send([‘Hello‘, None, ‘World‘])  # 写一段数据

5.    conn.close()  # 关闭

6. if __name__ == ‘__main__‘:

7.    parent_conn, child_conn = Pipe()  # 生成一个管道实例，parent_conn, child_conn管道的两头

8.    p = Process(target=ChildProcess, args=(child_conn,))

9.    p.start()

10.    print(parent_conn.recv())  # 收取消息

11.    p.join()

数据共享(Managers)

1. # _*_ coding:utf-8 _*_

2. # _*_ coding:utf-8 _*_

3. from multiprocessing import Process, Manager

4. import os

6. def ChildProcess(Dict, List):

7.    Dict[‘k1‘] = ‘v1‘

8.    Dict[‘k2‘] = ‘v2‘

9.    List.append(os.getpid())  # 获取子进程的PID

10.    print(List)  # 输出列表中的内容

12. if __name__ == ‘__main__‘:

13.    manager = Manager()  # 生成Manager对象

14.    Dict = manager.dict()  # 生成一个可以在多个进程之间传递共享的字典

15.    List = manager.list()  # 生成一个字典

17.    ProcessList = []  # 创建一个空列表，存放进程的对象，等待子进程执行用于

19.    for i in range(10):  # 生成是个子进程

20.        p = Process(target=ChildProcess, args=(Dict, List))  # 创建一个子进程

21.        p.start()  # 启动

22.        ProcessList.append(p)  # 把子进程添加到p_list列表中

24.    for res in ProcessList:  # 循环所有的子进程

25.        res.join()  # 等待执行完毕

26.    print(‘\n‘)

27.    print(Dict)

28.    print(List)

输出结果

1. C:\Python\Python35\python.exe E:/MyCodeProjects/多进程/s4.py

2. [5112]

3. [5112, 3448]

4. [5112, 3448, 4584]

5. [5112, 3448, 4584, 2128]

6. [5112, 3448, 4584, 2128, 11124]

7. [5112, 3448, 4584, 2128, 11124, 10628]

8. [5112, 3448, 4584, 2128, 11124, 10628, 5512]

9. [5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460]

10. [5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460, 10484]

11. [5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460, 10484, 6804]

14. {‘k1‘: ‘v1‘, ‘k2‘: ‘v2‘}

15. [5112, 3448, 4584, 2128, 11124, 10628, 5512, 10460, 10484, 6804]

17. Process finished with exit code 0

锁(Lock)

1. from multiprocessing import Process, Lock

3. def ChildProcess(l, i):

4.    l.acquire()  # 获取锁

5.    print(‘hello world‘, i)

6.    l.release()  # 释放锁

8. if __name__ == ‘__main__‘:

9.    lock = Lock()  # 生成Lock对象

10.    for num in range(10):

11.        Process(target=ChildProcess, args=(lock, num)).start()  # 创建并启动一个子进程

进程池

同一时间启动多少个进程

1. #!/use/bin/env python

2. # _*_ coding: utf-8 _*_

4. from multiprocessing import Pool

5. import time

7. def myFun(i):

8.    time.sleep(2)

9.    return i+100

11. def end_call(arg):

12.    print("end_call>>", arg)

14. p = Pool(5)  # 允许进程池内同时放入5个进程

15. for i in range(10):

16.    p.apply_async(func=myFun, args=(i,),callback=end_call) # # 平行执行,callback是主进程来调用

17.    # p.apply(func=Foo)  # 串行执行

19. print("end")

20. p.close()

21. p.join() # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

线程池

简单实现

1. #!/usr/bin/env python

2. # -*- coding:utf-8 -*-

3. import threading

4. import queue

5. import time

6. class MyThread:

7.    def __init__(self,max_num=10):

8.        self.queue = queue.Queue()

9.        for n in range(max_num):

10.            self.queue.put(threading.Thread)

11.    def get_thread(self):

12.        return self.queue.get()

13.    def put_thread(self):

14.        self.queue.put(threading.Thread)

15. pool = MyThread(5)

16. def RunThread(arg,pool):

17.    print(arg)

18.    time.sleep(2)

19.    pool.put_thread()

20. for n in range(30):

21.    thread = pool.get_thread()

22.    t = thread(target=RunThread, args=(n,pool,))

23.    t.start()

复杂版本

1. #!/usr/bin/env python

2. # -*- coding:utf-8 -*-

4. import queue

5. import threading

6. import contextlib

7. import time

9. StopEvent = object()

11. class ThreadPool(object):

13.    def __init__(self, max_num, max_task_num = None):

14.        if max_task_num:

15.            self.q = queue.Queue(max_task_num)

16.        else:

17.            self.q = queue.Queue()

18.        self.max_num = max_num

19.        self.cancel = False

20.        self.terminal = False

21.        self.generate_list = []

22.        self.free_list = []

24.    def run(self, func, args, callback=None):

25.        """

26.        线程池执行一个任务

27.        :param func: 任务函数

28.        :param args: 任务函数所需参数

29.        :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）

30.        :return: 如果线程池已经终止，则返回True否则None

31.        """

32.        if self.cancel:

34.            return

35.        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:

36.            self.generate_thread()

37.        w = (func, args, callback,)

38.        self.q.put(w)

40.    def generate_thread(self):

41.        """

42.        创建一个线程

43.        """

44.        t = threading.Thread(target=self.call)

45.        t.start()

47.    def call(self):

48.        """

49.        循环去获取任务函数并执行任务函数

50.        """

51.        current_thread = threading.currentThread()

52.        self.generate_list.append(current_thread)

54.        event = self.q.get()

55.        while event != StopEvent:

57.            func, arguments, callback = event

58.            try:

59.                result = func(*arguments)

60.                success = True

61.            except Exception as e:

62.                success = False

63.                result = None

65.            if callback is not None:

66.                try:

67.                    callback(success, result)

68.                except Exception as e:

69.                    pass

71.            with self.worker_state(self.free_list, current_thread):

72.                if self.terminal:

73.                    event = StopEvent

74.                else:

75.                    event = self.q.get()

76.        else:

78.            self.generate_list.remove(current_thread)

80.    def close(self):

81.        """

82.        执行完所有的任务后，所有线程停止

83.        """

84.        self.cancel = True

85.        full_size = len(self.generate_list)

86.        while full_size:

87.            self.q.put(StopEvent)

88.            full_size -= 1

90.    def terminate(self):

91.        """

92.        无论是否还有任务，终止线程

93.        """

94.        self.terminal = True

96.        while self.generate_list:

97.            self.q.put(StopEvent)

99.        self.q.queue.clear()

101.    @contextlib.contextmanager

102.    def worker_state(self, state_list, worker_thread):

103.        """

104.        用于记录线程中正在等待的线程数

105.        """

106.        state_list.append(worker_thread)

107.        try:

108.            yield

109.        finally:

110.            state_list.remove(worker_thread)

112. # How to use

114. pool = ThreadPool(5)

116. def callback(status, result):

117.    # status, execute action status

118.    # result, execute action return value

119.    pass

121. def action(i):

122.    print(i)

124. for i in range(30):

125.    ret = pool.run(action, (i,), callback)

127. time.sleep(5)

128. print(len(pool.generate_list), len(pool.free_list))

129. print(len(pool.generate_list), len(pool.free_list))

130. pool.close()

131. pool.terminate()

什么是IO密集型和CPU密集型？

IO密集型(I/O bound)

频繁网络传输、读取硬盘及其他IO设备称之为IO密集型，最简单的就是硬盘存取数据，IO操作并不会涉及到CPU。

计算密集型(CPU bound)

程序大部分在做计算、逻辑判断、循环导致cpu占用率很高的情况，称之为计算密集型，比如说python程序中执行了一段代码1+1，这就是在计算1+1的值

本文出自 “一盏烛光” 博客，谢绝转载！

原文：http://yw666.blog.51cto.com/11977292/1922769