「Python数据分析基础-3」Python的日期和时间及其处理的常用方法

off999 2024-10-14 12:09 43 浏览 0 评论

Python的线程以及线程间通信（队列）以及互斥锁的使用

一、什么是线程

如果需要同时处理多个任务，一种是可以在一个应用程序内使用多个进程，每个进程负责完成一部分工作;另一种将工作细分为多个任务的方法是使用一个进程内的多个线程。那么，什么是线程呢?

线程(Thread)是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。例如，对于视频播放器，显示视频用一个线程，播放音频用另一个线程。只有2个线程同时工作，我们才能正常观看画面和声音同步的视频。

举一个生活中的例子来更好地理解进程和线程的关系。一个进程就像一座房子，它是一个容器，有着相应的属性，如占地面积、卧室、厨房和卫生间等。房子本身并没有主动地做任何事情。而线程就是这座房子的居住者，他可以使用房子内每一个房间、做饭、洗澡等。

二、创建线程

由于线程是操作系统直接支持的执行单元，因此，高级语言（如Python、Java等）通常都内置多线程的支持。Python 的标准库提供了两个模块:_thread和 threading，_thread是低级模块，threading 是高级模块，对_thread进行了封装。绝大多数情况下，我们只需要使用threading这个高级模块。

1、使用threading模块创建线程

threading模块提供了一个Thread类来代表一个线程对象，语法如下:

Thread( [group [ , target [, name [, args [,kwargs]]]]])

Thread类的参数说明如下:

group :值为None，为以后版本而保留。
target:表示一个可调用对象，线程启动时，run()方法将调用此对象，默认值为None，表示不调用任何内容。
name表示当前线程名称，默认创建一个“Thread-N”格式的唯一名称。args :表示传递给target()函数的参数元组。
kwargs :表示传递给
target()函数的参数字典。

import threading,time

def process():
    for i in range(3):
        time.sleep(2)
        print("thread name is %s"%threading.current_thread().name)

if __name__=="__main__":
    print("----主线程开始----")
    #创建4个线程，存入列表
    threads = [threading.Thread(target=process) for i in range(4)]
    for t in threads:
        t.start()        #分别开启每个线程
    for t in threads:
        t.join()         #主线程阻塞，等待每个子线程结束
    print("----主线程结束----")

可以看出，线程执行的顺序是不确定的。

2、使用Thread子类创建线程

import threading,time
class SubThread(threading.Thread):
   def run(self):
       for i in range(3):
           time.sleep(1)
           #name属性保存的是当前线程的名字
           message = "子线程"+self.name+"执行，i="+str(i)
           print(message)

if __name__ == "__main__":
    print("----主线程开始----")
    #创建子线程
    t1 = SubThread()
    t2 = SubThread()
    #启动子线程
    t1.start()
    t2.start()
    #等待子线程结束
    t1.join()
    t2.join()
    print("----主线程结束----")

3、线程间通信

进程间不能直接共享信息，但是线程之间可以直接共享信息

from threading import Thread
import time

def th1():
    print("----子线程1开始----")
    global g_num
    g_num += 50
    print("g_num is %d"%g_num)
    print("----子线程1结束----")

def th2():
    time.sleep(1)
    print("----子线程2开始----")
    global g_num
    g_num -= 50
    print("g_num is %d"%g_num)
    print("----子线程2结束----")

g_num = 100

if __name__ == "__main__":
    print("----主线程开始----")
    print("g_num is %d"%g_num)
    #实例化线程
    t1 = Thread(target=th1)
    t2 = Thread(target=th2)
    #开始线程
    t1.start()
    t2.start()
    #线程等待，阻塞主线程
    t1.join()
    t2.join()
    print("----主线程结束----")

上述代码中，定义一个全局变量g_num，赋值为100，然后创建2个线程。一个线程g_num增加50，一个线程将g_num减少50。如果g_num的最终结果为100，则说明线程之间可以共享数据。

上述结果说明：

在一个进程内的所有线程共享全局变量，能够在不使用其他方式的前提下完成多线程之间的数据共享。

4、使用队列在线程间通信

我们知道multiprocessing模块的Queue队列可以实现进程间通信，同样在线程间，也可以使用Queue队列实现线程间通信。不同之处在于我们需要使用queue模块的Queue队列，而不是multiprocessing模块的Queue队列，但Queue的使用方法相同。

使用Queue在线程间通信通常应用于生产者消费者模式。==产生数据的模块称为生产者，而处理数据的模块称为消费者。在生产者与消费者之间的缓冲区称之为仓库。生产者负责往仓库运输商品，而消费者负责从仓库里取出商品，这就构成了生产者、消费者模式。==下面通过一个示例学习一下使用Queue在线程间通信。

定义一个生产者类Producer，定义一个消费者类Consumer。生成者生成5件产品，依次写入队列,而消费者依次从队列中取出产品，代码如下:

from queue import Queue
import queue
import random,threading,time

#生产者类
class Producer(threading.Thread):
    def __init__(self,name,queue):
        threading.Thread.__init__(self,name=name)
        self.data = queue
    def run(self):
        for i in range(5):
            print("生产者%s将产品%d加入队列!"%(self.getName(),i))
            self.data.put(i)
            time.sleep(random.random())
        print("生产者%s完成!"%self.getName())

#消费者类
class Consumer(threading.Thread):
    def __init__(self,name,queue):
        threading.Thread.__init__(self,name=name)
        self.data = queue
    def run(self):
        for i in range(5):
            val = self.data.get()
            print("消费者%s将产品%d从队列中取出!"%(self.getName(),val))
            time.sleep(random.random())
        print("消费者%s完成!"%self.getName()) 

if __name__ == "__main__":
    print("----主线程开始----")
    queue = Queue()                         #实例化队列
    producer = Producer("Producer",queue)   #实例化线程Producer，并传入队列作为参数
    consumer = Consumer("Consumer",queue)   #实例化线程Consumer，并传入队列作为参数
    producer.start()                        #启动线程
    consumer.start()
    producer.join()                         #等待producer线程结束
    consumer.join()                         #等待consumer线程结束
    print("----主线程结束----")

random.random()随机生成0~1之间的数

三、互斥锁

==由于线程可以对全局变量随意修改，这就可能造成多线程之间对全局变量的混乱操作。==依然以房子为例，当房子内只有一个居住者时（单线程)，他可以任意时刻使用任意一个房间，如厨房、卧室和卫生间等。但是，当这个房子有多个居住者时（多线程)，他就不能在任意时刻使用某些房间，如卫生间，否则就会造成混乱。

如何解决这个问题呢?一个防止他人进入的简单方法，就是门上加一把锁。先到的人锁上门，后到的人就在门口排队，等锁打开再进去。

这就是“互斥锁”(Mutual exclusion，缩写 Mutex)，防止多个线程同时读写某一块内存区域。互斥锁为资源引入一个状态:锁定和非锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改;直到该线程释放资源，将资源的状态变成“非锁定”时，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

1、使用互斥锁

在threading模块中使用Lock类可以方便地处理锁定。Lock类有2个方法: acquire()锁定和release()释放锁。示例用法如下:

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

语法如下:

acquire([blocking]):获取锁定，如果有必要，需要阻塞到锁定释放为止。如果提供blocking参数并将它设置为False，当无法获取锁定时将立即返回False;如果成功获取锁定则返回True。
release():释放一个锁定。当锁定处于未锁定状态时，或者从与原本调用acquire()方法的不同线程调用此方法,将出现错误。

下面，通过一个示例学习一下如何使用互斥锁。这里使用多线程和互斥锁模拟实现多人同时订购电影票的功能，假设电影院某个场次只有100张电影票，10个用户同时抢购该电影票。每售出一张，显示一次剩余的电影票张数。代码如下:

from threading import Thread 
from threading import Lock
import time

n = 100                             #共100张票
def task():
    global n                       
    mutex.acquire()                 #上锁
    # temp = n                        #赋值给临时变量
    # time.sleep(0.1)                 #休眠0.1秒
    # n = temp - 1                    #数量减1
    time.sleep(0.1)                 #休眠0.1秒
    n = n-1
    print("购买成功,剩余%d张电影票"%n)
    mutex.release()                 #释放锁 

if __name__ == "__main__":
    mutex = Lock()                  #实例化LOCK类
    t_list = []                     #初始化一个列表
    for i in range(10):
        t = Thread(target=task)     #实例化线程类
        t_list.append(t)            #将线程实例存入列表中
        t.start()                   #创建线程
    
    for t in t_list:
        t.join()                    #等待子线程结束

上述代码中，创建了10个线程，全部执行task()函数。为解决资源竞争问题，使用mutex.acquire()函数实现资源锁定，第一个获取资源的线程锁定后，其他线程等待mutex.release()解锁。所以每次只有一个线程执行task()函数。运行结果如图所示：

注意：