并发编程之多线程(实践)

Python多线程模块

threading模块介绍

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

Thread类介绍

threading模块有一个很重要的类叫做Thread类, Thread类是创建线程的类

Thread([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子线程中的任务（尚未启动）

强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

参数介绍：
1 group参数未使用，值始终为None
2 target表示调用对象，即子线程要执行的任务
3 args表示调用对象的位置参数元组，args=(1,2,'egon',)
4 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
5 name为子线程的名称

方法介绍:
1 t.start()：启动线程，并调用该子线程中的p.run() 
2 t.run(): 线程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法  
3 t.join([timeout]):主线程等待t终止（强调：是主线程处于等的状态，而t是处于运行的状态）。timeout是可选的超时时间，需要强调的是，t.join只能join住start开启的线程，而不能join住run开启的线程 

属性介绍:
1 t.daemon：默认值为False，如果设为True，代表t为后台运行的守护进程，当t的父进程终止时，t也随之终止，并且设定为True后，t不能创建自己的新线程，必须在t.start()之前设置

创建线程

基于Thread类创建线程

from threading import Thread
import time


def task(name):
    print('%s is running' % name)
    time.sleep(3)
    print('%s is over' % name)


# 开启线程不需要在main下面执行代码, 直接书写就可以了, 但是还是习惯性的将启动命令卸载main下面
if __name__ == '__main__':
    t = Thread(target=task, args=('featherwit',))
    t.start()  # 创建线程的开销非常小, 几乎是代码一执行线程就创建了
    print('主线程...')

继承Thread类创建线程

class MyThread(Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self) -> None:
        print('%s is runing' % self.name)
        time.sleep(3)
        print('%s is over' % self.name)


if __name__ == '__main__':
    t = MyThread('featherwit')
    t.start()
    print('主线程...')

 

案例: 多线程实现socket并发

服务端代码:

import socket
from threading import Thread
from multiprocessing import Process

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 括号内不加参数默认就是TCP协议
server.bind(('127.0.0.1', 8080))
server.listen(5)


# 将服务的代码封装成一个函数
def talk(conn):
    # 通信循环
    while True:
        try:
            data = conn.recv(1024)
            if len(data) == 0: break
            print(data.decode('utf-8'))
            conn.send(data.upper())
        except ConnectionResetError as e:
            print(e)
            break

    conn.close()


# 链接循环
while True:
    conn, addr = server.accept()
    t = Thread(target=talk, args=(conn,))
    t.start()

客户端代码:

import socket

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

while True:
    try:
        client.send(b'hello world')
        data = client.recv(1024)
        print(data.decode('utf-8'))
    except Exception as e:
        print(e)
        break

client.close()

Thread类的其他方法

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

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

案例:

from threading import Thread, current_thread, active_count
import time
import os


def task(n):
    print('hello world')
    print('子线程...: PID', os.getpid)
    print('获取子线程的名字...: name', current_thread().name)
    time.sleep(n)


if __name__ == '__main__':
    t1 = Thread(target=task, args=(1,))
    t2 = Thread(target=task, args=(2,))
    t1.start()
    t2.start()
    t1.join()
    print('主线程...: PID', os.getpid)
    print('统计当前正在活跃的线程数: ', active_count())
    print('获取主线程的名字...: name', current_thread().name)

同一进程下的线程数据共享

from threading import Thread

money = 100


def task():
    global money
    money = 666


if __name__ == '__main__':
    t = Thread(target=task)
    t.start()
    t.join()
    print(money)

守护线程

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

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

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

from threading import Thread
import time


def task(name):
    print('%s is running' % name)
    time.sleep(1)
    print('%s is over' % name)


if __name__ == '__main__':
    t = Thread(target=task, args=('featherwit', ))
    t.setDaemon(True)
    t.start()
    print('主线程...')

join方法

主线程运行结束之后不会立即结束, 会等待所有其他非守护线程结束才会结束因为主线程的结束意味着所在的进程的结束

from threading import Thread
import time


def task(name):
    print('%s in running' % name)
    time.sleep(3)
    print('%s in over' % name)


if __name__ == '__main__':
    t = Thread(target=task, args=('featherwit',))
    t.daemon = True
    t.start()
    t.join()  # 主线程等待子线程运行结束再执行
    print('主线程')

锁

互斥锁

不加锁, 线程间资源竞争的问题

from threading import Thread
import time

money = 100


def task():
    global money
    tmp = money
    time.sleep(1)
    money = tmp - 1


if __name__ == '__main__':
    t_list = []
    for i in range(1, 101):
        t = Thread(target=task)
        t.start()
        t_list.append(t)

    print(money)

此时, money的结果可能为99。加锁, 可以解决资源之间的竞争问题

from threading import Thread, Lock
import time

money = 100


def task(mutex):
    global money
    mutex.acquire()
    tmp = money
    # time.sleep(1)
    money = tmp - 1
    mutex.release()


if __name__ == '__main__':
    mutex = Lock()
    t_list = []
    for i in range(1, 101):
        t = Thread(target=task, args=(mutex, ))
        t.start()
        t_list.append(t)

    for t in t_list:
        t.join()

    print(money)

线程由原来的并发执行变成串行，牺牲了执行效率保证了数据安全。

那么, 加锁和join()都能将线程变成串行执行, 这两者之间有什么区别呢?

在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是start后立即join, 任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的

单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.

死锁与递归锁

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

from threading import Thread, Lock
import time

mutexA = Lock()
mutexB = Lock()


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

    def func1(self):
        mutexA.acquire()
        print('%s 抢到A锁' % self.name)  # 获取当前线程名
        mutexB.acquire()
        print('%s 抢到B锁' % self.name)
        mutexB.release()
        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('%s 抢到B锁' % self.name)
        time.sleep(2)
        mutexA.acquire()
        print('%s 抢到A锁' % self.name)
        mutexA.release()
        mutexB.release()


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

在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release, 即counter为0的时候, 其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁。

from threading import Thread, RLock
import time

mutexA = mutexB = RLock()


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

    def func1(self):
        mutexA.acquire()
        print('%s 抢到A锁' % self.name)  # 获取当前线程名
        mutexB.acquire()
        print('%s 抢到B锁' % self.name)
        mutexB.release()
        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('%s 抢到B锁' % self.name)
        time.sleep(2)
        mutexA.acquire()
        print('%s 抢到A锁' % self.name)
        mutexA.release()
        mutexB.release()


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

 

线程高级操作

信号量

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

from threading import Thread, Semaphore
import time
import random

sm = Semaphore(5)


def task(name):
    sm.acquire()
    print('%s 正拿着锁' % name)
    time.sleep(random.randint(1, 5))
    sm.release()


if __name__ == '__main__':
    for i in range(20):
        t = Thread(target=task, args=('%s号' % i, ))
        t.start()

Event事件

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

from threading import Thread, Event
import time

event = Event()


def light():
    print('红灯亮着的')
    time.sleep(3)
    print('绿灯亮着的')
    # 告诉等待红灯的人可以走了
    event.set()


def car(name):
    print('%s 车正在等红灯' % name)
    # 别人通知我说不用等了
    event.wait()  # 等待别人给你发消息
    print('%s 车可以走了' % name)


if __name__ == '__main__':
    t = Thread(target=light)
    t.start()

    for i in range(20):
        t = Thread(target=car, args=('兰博基尼%s号' % i, ))
        t.start()

定时器

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

from threading import Timer


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

t = Timer(1, hello)
t.start()

线程队列

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

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

import queue

# 1. Queue, 先进先出
"""
q = queue.Queue(3)
q.put(1)
q.get()
q.get_nowait()
q.get(timeout=3)
q.full()
q.empty()
"""

# 2. LifoQueue, 后进先出
q = queue.LifoQueue(3)
q.put(1)
q.put(2)
q.put(3)
print(q.get())

# 3. 优先级Queue, 你可以给放入队列中的数据设置进出的优先级
q = queue.PriorityQueue(3)
q.put((10, '111'))
q.put((100, '222'))
q.put((0, '444'))
print(q.get())
# put括号内放一个元组, 第一个放数字表示优先级, 需要注意的是数字越小, 优先级越高