python3多线程threading、_thread、Queue

Dear 丶 2022-04-05 14:12 443阅读 0赞

# 1.多线程 #

> 多线程**类似于同时执行多个不同程序**，多线程运行有如下**优点**：
> 
> ①可将占据时间长的程序中的任务放到后台去处理
> 
> ②用户界面更加吸引人，如：用户点击某按钮并触发事件，可弹出进度条显示处理的进度
> 
> ③在一些等待的任务实现上如用户输入、文件读写和网络收发数据等，线程就有了用武之地。在此种情况下可释放一些珍贵的资源如内存占用等。

# 2.\_thread模块 #

> \_thread模块提供了低级别的、原始的线程及一个简单的锁，它相比于threading模块的功能还是比较有限的。
> 
> \_thread.**start\_new\_thread**(**function**, **args**, **kwargs**)
> 
> \_thread.**start\_new**(function, args, kwargs) --> 过时的同意表达
> 
> 参数：
> 
> **function**: 线程函数
> 
> **args**: 传递给function的参数，必须为tuple类型
> 
> **kwargs**: 可选参数

import _thread
    import time
    
    
    def print_time(threadName, delay):
        """定义线程函数"""
        count = 0
        while count < 5:
            time.sleep(delay)
            count += 1
            print("%s: %s [count:%s]" % (threadName, time.ctime(time.time()), count))
    
    
    # 创建两个线程
    try:
        _thread.start_new_thread(print_time, ("Thread-1", 2))
        _thread.start_new_thread(print_time, ("Thread-2", 4))
        _thread.start_new(print_time, ("Thread-3", 8))
    except:
        print("Error: 无法其中进程")
    
    
    while 1:
        """长时间占用任务"""
        pass

# 3.threading模块 #

> threading模块除了包含\_thread模块中的所有方法外，还提供了如下方法：
> 
> **threading.currentThread()**: 返回当前线程的变量
> 
> **threading.enumerate()**: 返回一个包含正在运行的线程的列表
> 
> **threading.activeCount()**: 返回正在运行的线程数量 == len(threading.enumerate())

## 3.1 创建线程的两种方式 ##

> 方式①：
> 
> 将要执行的方法作为参数传给Thread的构造方法
> 
> 方法②：
> 
> 从Thread继承，并重写run()  \# 3.2 中代码已经给出

def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None):
        pass
    
    
    # 参数：
    # 1.group: should be None; reserved for future extension when a ThreadGroup class is 
    # implemented.
    
    # 2.target: is the callable object to be invoked by the run() method. Defaults to None, 
    # meaning nothing is called.
    
    # 3.name: is the thread name. By default, a unique name is constructed of the form "Thread-
    # N" where N is a small decimal number.
    
    # 4.args: is the argument tuple for the target invocation. Defaults to ().
    
    # 5.kwargs: is a dictionary of keyword arguments for the target invocation. Defaults to {}
    
    # If a subclass overrides the constructor, it must make sure to invoke
    # the base class constructor (Thread.__init__()) before doing anything
    # else to the thread.

import threading
    import time
    
    
    #方法一：将要执行的方法作为参数传给Thread的构造方法
    def action(arg, els):
        time.sleep(1)
        print('the arg is:(%s, %s)\r' % (arg, els))
    
    
    for i in range(4):
        t =threading.Thread(target=action, name="Tom"+str(i), args=(i, i))
        t.start()
        print(t.getName())
    
    print('main thread end!')

## 3.2 thread类 ##

> **run()**: 线程启动之后，执行预先下入的程序代码
> 
> **start()**: 启动线程
> 
> **join(\[time\])**: 等待至线程终止  \# 经测试无用
> 
> **isAlive()**: 返回线程是否是活动的
> 
> **getName()**: 返回线程名称
> 
> **setName()**: 设置线程名称

import threading
    import time
    
    
    count = 0
    class MyThread(threading.Thread):
        def __init__(self, threadName):
            super(MyThread, self).__init__(name=threadName)
    
        def run(self):
            # 声明count为全局变量
            print("开始线程:" + self.name)
            global count
            for i in range(2):
                count += 1
                time.sleep(0.3)
                print(self.getName(), count)
            print("结束线程:" + self.name)
    
    
    if "__main__" == __name__:
        thread1 = MyThread("MyThreadName:" + str(1))
        thread2 = MyThread("MyThreadName:" + str(2))
        thread1.start()
        print("*" * 100)
        
        # 判断thread1线程是否存活
        print(thread1.isAlive())  # True
        thread1.join()
        print(thread1.isAlive())  # False
        thread2.start()
        thread2.join()

## 3.3 lock和Rlock类（线程同步） ##

> 若多个线程共同对某个数据修改，则可能出现不能预料的结果，为保证数据的正确性，需要对多个线程进行同步。使用Threading的lock类和Rlock类可以实现简单的同步，这两个类所创建的对象都有acquire()和release()，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到**acquire()**和**release()**方法之间。
> 
> 锁有两种状态——锁定和未锁定。每当一个线程比如“line1”要**访问共享数据时，必须先获得锁**；如果已经没有别的线程比如“line2”获得锁，那么就让线程“line1”同步阻塞；等到线程“line2”完毕并释放共享数据的锁之后，再让线程“line1”继续对数据进行操作。

import threading
    import time
    
    
    class MyThread(threading.Thread):
        def __init__(self,threadName):
            # def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None)
            super(MyThread, self).__init__(name=threadName)
    
        def run(self):
            print("开启线程：", self.getName())
    
            # 获得锁，用于线程同步
            threadLock.acquire()
            print_time(self.getName())
            # 释放锁
            threadLock.release()
    
    
    def print_time(threadName):
        counter = 3
        while counter:
            time.sleep(3)
            print("%s: %s" % (threadName, time.ctime(time.time())))
            counter -= 1
    
    
    if "__main__" == __name__:
        threadLock = threading.Lock()
        threads = []
    
        thread1 = MyThread("name1")
        thread2 = MyThread("name2")
    
        thread1.start()
        thread2.start()
    
        # 添加线程到线程列表
        threads.append(thread1)
        threads.append(thread2)
    # 等待所有线程完成
        for t in threads:
            t.join()
        print("退出主线程")

## 3.4 join类 ##

> join类是threading模块中**用于堵塞当前主线程的类**，其作用是阻止全部的线程继续运行，直到被调用的线程执行完毕或者超时。

# 4. 线程优先级队列（Queue） #

> python中Queue模块中提供同步的、线程安全的队列类，包括**FIFO（先入先出）**、**LIFO（后入先出）队列**和**PriorityQueue（优先级队列）**。这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程的同步。
> 
> Queue模块中常用方法：
> 
> **Queue.qsize()**: 返回队列大小
> 
> **Queue.empty()**: 判断队列是否为empty
> 
> **Queue.full()**: 判断队列是否为满
> 
> Queue.maxsize
> 
> **Queue.get(block, timeout)**: 获取队列，timeout等待时间
> 
> Queue.get(item, False) == Queue.get\_nowait(item)
> 
> **Queue.put(item，timeout)**：写入队列，timeout等待时间
> 
> Queue.put(item, False) == Queue.get\_nowait(item)
> 
> **Queue.task\_done()**：在完成一项工作之后，Queue.task\_done()函数向任务完成的队列发一个信号
> 
> **Queue.join()**：实际上意味着等到队列为空，再执行别的操作

import threading
    import queue
    import time
    
    
    exitFlag = 0
    
    
    class myThread(threading.Thread):
        def __init__(self, threadID, name, q):
            super(myThread, self).__init__(name=name)
            self.threadID = threadID
            self.q = q
    
        def run(self):
            print("start threading ...[%s]" % self.getName())
            process_data(self.getName(), self.q)
            print("end threading... [%s]" % self.getName())
    
    
    def process_data(threadName, q):
        while not exitFlag:
            # 获取锁
            queueLock.acquire()
            if not workQueue.empty():
                data = q.get()
                # 释放锁
                queueLock.release()
                print("%s processing %s" % (threadName, data))
            else:
                # 释放锁
                queueLock.release()
            time.sleep(1)
    
    
    threadList = ["Thread-1", "Thread-2", "Thread-3"]
    nameList = ["One", "Two", "Three", "Four", "Five"]
    queueLock = threading.Lock()
    workQueue = queue.Queue(10)  # 创建大小为10的队列
    threads = []
    threadID = 1
    
    # 创建新线程
    for tName in threadList:
        thread = myThread(threadID, tName, workQueue)
        thread.start()
        threads.append(thread)
        threadID += 1
    
    # 填充队列
    queueLock.acquire()
    for word in nameList:
        workQueue.put(word)
    queueLock.release()
    
    # 等待队列清空
    while not workQueue.empty():
        pass
    
    # 通知线程是时候退出
    exitFlag = 1
    
    # 等待所有线程完成
    for t in threads:
        t.join()
    print("退出主线程")