首页 > > 网络编程 > 其它 >

多任务—线程

2019-04-11 10:40:45来源：博客园阅读 ()

一、多任务的概念

什么叫“多任务”呢？简单地说，就是操作系统可以同时运行多个任务。打个比方，你一边在用浏览器上网，一边在听MP3，一边在用Word赶作业，这就是多任务，至少同时有3个任务正在运行。还有很多任务悄悄地在后台同时运行着，只是桌面上没有显示而已。

现在，多核CPU已经非常普及了，但是，即使过去的单核CPU，也可以执行多任务。由于CPU执行代码都是顺序执行的，那么，单核CPU是怎么执行多任务的呢？

答案就是操作系统轮流让各个任务交替执行，任务1执行0.01秒，切换到任务2，任务2执行0.01秒，再切换到任务3，执行0.01秒……这样反复执行下去。表面上看，每个任务都是交替执行的，但是，由于CPU的执行速度实在是太快了，我们感觉就像所有任务都在同时执行一样。

真正的并行执行多任务只能在多核CPU上实现，但是，由于任务数量远远多于CPU的核心数量，所以，操作系统也会自动把很多任务轮流调度到每个核心上执行。

注意：

并发：指的是任务数多余cpu核数，通过操作系统的各种任务调度算法，实现用多个任务“一起”执行（实际上总有一些任务不在执行，因为切换任务的速度相当快，看上去一起执行而已）
并行：指的是任务数小于等于cpu核数，即任务真的是一起执行的

二、线程

python的thread模块是比较底层的模块，python的threading模块是对thread做了一些包装的，可以更加方便的被使用

1. 使用threading模块

单线程执行

 import time
 
 
 def single_threading():
     print("年复一年")
     time.sleep(1)
 
 if __name__ == '__main__':
     for i in range(5):
         single_threading()

多线程执行

 import threading
 import time
 
 
 def multi_threading():
     print("年复一年")
     time.sleep(1)
 
 if __name__ == '__main__':
     for i in range(5):
        t = threading.Thread(target=multi_threading)
        t.start()

说明

可以明显看出使用了多线程并发的操作，花费时间要短很多
当调用start()时，才会真正的创建线程，并且开始执行

2. 主线程会等待所有的子线程结束后才结束

 import time
 import threading
 
 
 def sing():
     for i in range(3):
         print("singing.....%d" % i)
         time.sleep(1)
 
 def dance():
     for i in range(3):
         print("dancing....%d" % i)
         time.sleep(1)
 
 
 if __name__ == "__main__":
     print("start....%s" % time.ctime())
     t1 = threading.Thread(target=sing)
     t2 = threading.Thread(target=dance)
     t1.start()
     t2.start()
     time.sleep(5)
     print("end.....%s" % time.ctime())

View Code

3. 查看线程数量

import time
 import threading
 
 
 def sing():
     for i in range(3):
         print("singing.....%d" % i)
         time.sleep(1)
 
 def dance():
     for i in range(3):
         print("dancing....%d" % i)
         time.sleep(1)
 
 
 if __name__ == "__main__":
     print("start....%s" % time.ctime())
     t1 = threading.Thread(target=sing)
     t2 = threading.Thread(target=dance)
     t1.start()
     t2.start()
     while True:
         length = len(threading.enumerate())
         print("当前线程数:%d" % length)
         if length <= 1:
             break
         time.sleep(0.5)

View Code

三、线程-继承

1. 线程执行代码的封装

通过上一小节，能够看出，通过使用threading模块能完成多任务的程序开发，为了让每个线程的封装性更完美，所以使用threading模块时，往往会定义一个新的子类class，只要继承threading.Thread就可以了，然后重写run方法

示例如下：

 import threading
 import time
 
 
 class MyThread(threading.Thread):
 
     def run(self):
         for i in range(3):
             print("I'm "+self.name+' @ '+str(i))    # name属性中保存的是当前线程的名字
             time.sleep(1)
 
 
 if __name__ == "__main__":
     t = MyThread()
     t.start()

View Code

说明

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

2. 线程的执行顺序

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

View Code

说明

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

3. 总结

每个线程默认有一个名字，尽管上面的例子中没有指定线程对象的name，但是python会自动为线程指定一个名字。
当线程的run()方法结束时该线程完成。
无法控制线程调度程序，但可以通过别的方式来影响线程调度的方式。

四、多线程-共享全局变量

多线程共享全局变量

from threading import Thread
 import time
 
 
 g_num = 10
 
 def worker1():
     global g_num
     for i in range(3):
         g_num += 1
 
     print("in woker1 g_num: %d" % g_num)
 
 
 def worker2():
     global g_num
     print("in worker2 g_num: %d" % g_num)
 
 
 print("线程执行之前 g_num: %d" % g_num)
 
 t1 = Thread(target=worker1)
 t1.start()
 time.sleep(1)   # 保证worker1执行完
 
 t2 = Thread(target=worker2)
 t2.start()

View Code

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

from threading import Thread
 import time
 
 
 def worker1(nums):
     nums.append(44)
     print("in worker1", nums)
 
 def worker2(nums):
     print("in worker2", nums)
 
 
 num_list = [11,22,33]
 
 t1 = Thread(target=worker1, args=(num_list,))
 t1.start()
 
 time.sleep(1)
 
 t2 = Thread(target=worker2, args=(num_list,))
 t2.start()

View Code

总结：

在一个进程内的所有线程共享全局变量，很方便在多个线程间共享数据
缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱（即线程非安全）

多线程-共享全局变量问题

假设两个线程t1和t2都要对全局变量g_num(默认是0)进行加1运算，t1和t2都各对g_num加10次，g_num的最终的结果应该为20。

但是由于是多线程同时操作，有可能出现下面情况：

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

测试1：

 import threading
 import time
 
 
 g_num = 0
 
 def worker1(num):
     global g_num
     for i in range(num):
         g_num += 1
 
     print("in woker1 g_num: %d" % g_num)
 
 
 def worker2(num):
     global g_num
     for i in range(num):
         g_num += 1
     print("in worker2 g_num: %d" % g_num)
 
 
 t1 = threading.Thread(target=worker1, args=(100,))
 t1.start()
 
 t2 = threading.Thread(target=worker2, args=(100,))
 t2.start()
 
 if len(threading.enumerate()) > 1:
     time.sleep(1)
 
 
 print("两个线程对一个全局变量执行结果 g_num: %d" % g_num)

View Code

运行结果：

in woker1 g_num: 100
in worker2 g_num: 200
两个线程对一个全局变量执行结果 g_num: 200

测试2：

 import threading
 import time
 
 
 g_num = 0
 
 def worker1(num):
     global g_num
     for i in range(num):
         g_num += 1
 
     print("in woker1 g_num: %d" % g_num)
 
 
 def worker2(num):
     global g_num
     for i in range(num):
         g_num += 1
     print("in worker2 g_num: %d" % g_num)
 
 
 t1 = threading.Thread(target=worker1, args=(1000000,))
 t1.start()
 
 t2 = threading.Thread(target=worker2, args=(1000000,))
 t2.start()
 
 if len(threading.enumerate()) > 1:
     time.sleep(1)
 
 
 print("两个线程对一个全局变量执行结果 g_num: %d" % g_num)

View Code

运行结果：

in worker2 g_num: 1202086
in woker1 g_num: 1185129
两个线程对一个全局变量执行结果 g_num: 1185129

结论

如果多个线程同时对同一个全局变量操作，会出现资源竞争问题，从而数据结果会不正确

五、同步

同步的概念

同步就是协同步调，按预定的先后次序进行运行。如:你说完，我再说。

"同"字从字面上容易理解为一起动作

其实不是，"同"字应是指协同、协助、互相配合。

如进程、线程同步，可理解为进程或线程A和B一块配合，A执行到一定程度时要依靠B的某个结果，于是停下来，示意B运行;B执行，再将结果给A;A再继续操作。

解决线程同时修改全局变量的方式

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

思路，如下:

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

六、互斥锁

互斥锁概念

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

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

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

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

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

# 创建锁
mutex = threading.Lock()

# 锁定
mutex.acquire()

# 释放
mutex.release()

注意：

如果这个锁之前是没有上锁的，那么acquire不会堵塞
如果在调用acquire对这个锁上锁之前它已经被其他线程上了锁，那么此时acquire会堵塞，直到这个锁被解锁为止

使用互斥锁完成2个线程对同一个全局变量各加100万次的操作

 import threading
 import time
 
 
 g_num = 0
 
 def worker1(num):
     global g_num
     for i in range(num):
         mutex.acquire() # 上锁
         g_num += 1
         mutex.release() # 解锁
 
     print("in woker1 g_num: %d" % g_num)
 
 
 def worker2(num):
     global g_num
     for i in range(num):
         mutex.acquire() # 上锁
         g_num += 1
         mutex.release() # 解锁
     print("in worker2 g_num: %d" % g_num)
 
 
 # 创建互斥锁, 默认是为上锁状态
 mutex = threading.Lock()
 
 t1 = threading.Thread(target=worker1, args=(1000000,))
 t1.start()
 
 t2 = threading.Thread(target=worker2, args=(1000000,))
 t2.start()
 
 if len(threading.enumerate()) > 1:
     time.sleep(1)
 
 
 print("两个线程对一个全局变量执行结果 g_num: %d" % g_num)

View Code

结果：

in woker1 g_num: 1987171
in worker2 g_num: 2000000
两个线程对一个全局变量执行结果 g_num: 2000000

上锁解锁过程

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

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

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

总结

锁的好处：

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

锁的坏处：

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

七、死锁

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

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

 import threading
 import time
 
 
 class MyThread1(threading.Thread):
 
     def run(self):
         mutexA.acquire()    # mutexA上锁
         # mutexA上锁后，延时1秒，等待另外那个线程 把mutexB上锁
         print(self.name+'----do1---up----')
         time.sleep(1)
 
         # 此时会堵塞，因为这个mutexB已经被另外的线程抢先上锁了
         mutexB.acquire()
         print(self.name+'----do1---down----')
         mutexB.release()
 
         # mutexA解锁
         mutexA.release()
 
 
 class MyThread2(threading.Thread):
 
     def run(self):
         # mutexB上锁
         mutexB.acquire()
 
         # mutexB上锁后，延时1秒，等待另外那个线程 把mutexA上锁
         print(self.name+'----do2---up----')
         time.sleep(1)
 
         # 此时会堵塞，因为这个mutexA已经被另外的线程抢先上锁了
         mutexA.acquire()
         print(self.name+'----do2---down----')
         mutexA.release()
 
         # 对mutexB解锁
         mutexB.release()
 
 
 mutexA = threading.Lock()
 mutexB = threading.Lock()
 
 if __name__ == "__main__":
     t1 = MyThread1()
     t2 = MyThread2()
     t1.start()
     t2.start()

View Code

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

避免死锁

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

递归锁 解决死锁

from threading import 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()

Thread(target=eat1,args=('房书安',)).start()
Thread(target=eat2,args=('郭长达',)).start()
Thread(target=eat1,args=('朱亮',)).start()
Thread(target=eat2,args=('梅良祖',)).start()

View Code

八、udp多线程聊天器

import threading
 import socket
 
 
 def send_msg(udp_socket):
     while True:
         # 输入对方的地址
         addr_ip = input("对方的ip：")
         # 输入对方的端口
         addr_port = int(input("对方的端口："))
         # 输入信息
         send_data = input("输入消息：")
         # 发送
         udp_socket.sendto(send_data.encode('utf-8'),(addr_ip, addr_port))
 
 
 def recv_msg(udp_socket):
     while True:
         # 接受数据
         recv_data = udp_socket.recvfrom(1024)
         print("%s-->: %s" % (recv_data[1], recv_data[0].decode('utf-8')))
 
 
 if __name__ == "__main__":
     # 创建套接字
     udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     # 绑定
     udp_socket.bind(('192.168.1.1',8010))
     # 创建发送信息线程
     td_send = threading.Thread(target=send_msg, args=(udp_socket,))
     # 创建收信息线程
     td_recv = threading.Thread(target=recv_msg, args=(udp_socket,))
 
     td_send.start()
     td_recv.start()

View Code

九、守护线程

无论是进程还是线程，都遵循：守护xx会等待主xx运行完毕后被销毁。需要强调的是：运行完毕并非终止运行

1.对主进程来说，运行完毕指的是主进程代码运行完毕

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

详细解释：

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

 from threading import Thread
 import time
 
 def func1():
     time.sleep(2)
     print("end func1")
 
 
 
 if __name__ == "__main__":
     t1 = Thread(target=func1)
     t1.daemon = True
     t1.start()
     print("__main__")
     print(t1.is_alive())

# 执行结果：
__main__
True

例一

 from threading import Thread
 import time
 
 def func1():
     print("func1")
     time.sleep(1)
     print("end func1")
 
 def func2():
     print("func2")
     time.sleep(3)
     print("end func2")
 
 
 if __name__ == "__main__":
     t1 = Thread(target=func1)
     t2 = Thread(target=func2)
     t1.daemon = True
     t1.start()
     t2.start()
     print("__main__")


# 执行结果：
func1
func2
__main__
end func1
end func2

例二

十、线程队列

 import queue
 
 
 q = queue.Queue()
 
 q.put(1)
 q.put(2)
 q.put(3)
 q.put(4)
 
 
 print(q.get())
 print(q.get())
 print(q.get())
 print(q.get())

# 执行结果
1
2
3
4

View Code

GIL（全局解释器锁）

GIL面试题如下：

　　描述Python GIL的概念，以及它对python多线程的影响？编写一个多线程抓取网页的程序，并阐明多线程抓取程序是否可比单线程性能有提升，并解释原因。

Guido的声明：http://www.artima.com/forums/flat.jsp?forum=106&thread=214235

he language doesn't require the GIL -- it's only the CPython virtual machine that has historically been unable to shed it.

参考答案:

Python语言和GIL没有半毛钱关系。仅仅是由于历史原因在Cpython虚拟机(解释器)，难以移除GIL。
GIL：全局解释器锁。每个线程在执行的过程都需要先获取GIL，保证同一时刻只有一个线程可以执行代码。
线程释放GIL锁的情况：在IO操作等可能会引起阻塞的system call之前,可以暂时释放GIL,但在执行完毕后,必须重新获取GIL Python 3.x使用计时器（执行时间达到阈值后，当前线程释放GIL）或Python 2.x，tickets计数达到100
Python使用多进程是可以利用多核的CPU资源的。
多线程爬取比单线程性能有提升，因为遇到IO阻塞会自动释放GIL锁