python学习之进程线程学习二

Posted 2020-09-26

一、线程介绍
处理线程的模块是threading,multiprocessing模块处理方式跟threading相似

开启线程的两种方式:
例子:
from threading import Thread
from multiprocessing import Process

def work(name):
    print(‘%s say hello‘ %name)

if __name__ == ‘__main__‘:
    t = Thread(target=work, args=(‘hyh‘,))
    t.start()
    print(‘主线程‘)
    
class Work(Thread):
    def __init__(self,name):
        super().__init__()
        self.name = name

    def run(self):
        print(‘%s say hello‘ %self.name)

if __name__ == ‘__main__‘:
    t = Work(‘hyh‘)
    t.start()
    print(‘主线程‘)
    
二、线程方法
queue方法
例子:
import queue

q = queue.Queue(3)    #先进先出
q.put(1)
q.put(‘hyh‘)
q.put([1,2,3,4])

print(q.get())
print(q.get())
print(q.get())

q = queue.LifoQueue()    #后进先出
q.put(1)
q.put(‘hyh‘)
q.put([1,2,3,4])

print(q.get())
print(q.get())
print(q.get())

q = queue.PriorityQueue()    #优先级，数字越小优先级越高
q.put((10, ‘a‘))
q.put((9, ‘b‘))
q.put((11, ‘c‘))

print(q.get())
print(q.get())
print(q.get())

线程其他方法
例子:
import time
from threading import Thread
import threading

def work():
    time.sleep(2)
    print(‘%s say hello‘ %(threading.current_thread().getName()))

if __name__ == ‘__main__‘:
    t = Thread(target=work)
    t.setDaemon(True)    #设置成守护线程
    t.start()
    t.join()
    print(threading.enumerate())    #当前活跃的线程对象，是一个列表形式
    print(threading.active_count()) #当前活跃的线程数目
    print(‘主线程‘, threading.current_thread().getName())  #线程名字
    
三、python全局解释器锁GIL
python同一进程的线程利用不了多核优势，因为一个线程运行时获取GIL锁，等到运行结束释放GIL，
其它线程才能申请GIL

现在的计算机基本上都是多核，python对于计算密集型的任务开多线程的效率并不能带来多大性能上
的提升，甚至不如串行(没有大量切换)，但是，对于IO密集型的任务效率还是有显著提升的

例子:
计算密集型
from threading import Thread
from multiprocessing import Process
import os
import time

def work():
    res = 0
    for i in range(1000000):
        res += i

if __name__ == ‘__main__‘:
    t_l = []
    start_time = time.time()
    for i in range(300):
        t = Thread(target=work)
        t_l.append(t)
        t.start()

    for i in t_l:
        i.join()

    stop_time = time.time()
    print(‘run time is %s‘ %(stop_time - start_time))
    print(‘主线程‘)

IO密集型    
from threading import Thread
from multiprocessing import Process
import time
import os

def work():
    time.sleep(2)
    print(os.getpid())

if __name__ == ‘__main__‘:
    t_l = []
    start_time = time.time()
    for i in range(1000):
        t = Thread(target=work)
        t_l.append(t)
        t.start()

    for t in t_l:
        t.join()
    stop_time = time.time()
    print(‘run time is %s‘ %(stop_time - start_time))

线程锁Lock
import threading

R=threading.Lock()

R.acquire()
‘‘‘
对公共数据的操作
‘‘‘
R.release()


死锁
例子:
from threading import Thread,Lock
import time
mutexA = Lock()
mutexB = Lock()

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

    def func1(self):
        mutexA.acquire()
        print(‘\033[41m%s 拿到A锁\033[0m‘ %self.name)

        mutexB.acquire()
        print(‘\033[42m%s 拿到B锁\033[0m‘ %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print(‘\033[43m%s 拿到B锁\033[0m‘ %self.name)
        time.sleep(2)

        mutexA.acquire()
        print(‘\033[44m%s拿到A锁\033[0m‘ %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == ‘__main__‘:
    for i in range(10):
        t = MyThread()
        t.start()
        
输出结果:
Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-2 拿到A锁
卡住。。。


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

from threading import Thread,RLock
import time
mutex = RLock()


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

    def func1(self):
        mutex.acquire()
        print(‘\033[41m%s 拿到A锁\033[0m‘ %self.name)

        mutex.acquire()
        print(‘\033[42m%s 拿到B锁\033[0m‘ %self.name)
        mutex.release()

        mutex.release()

    def func2(self):
        mutex.acquire()
        print(‘\033[43m%s 拿到B锁\033[0m‘ %self.name)
        time.sleep(2)

        mutex.acquire()
        print(‘\033[44m%s拿到A锁\033[0m‘ %self.name)
        mutex.release()

        mutex.release()

if __name__ == ‘__main__‘:
    for i in range(10):
        t = MyThread()
        t.start()
        
信号量Semahpore
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()
例子:
import threading
import time

semaphore = threading.Semaphore(5)

def func():
    if semaphore.acquire():
        print(threading.current_thread().getName() + ‘ get spmaphore‘)
        time.sleep(2)
        semaphore.release()

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

event.wait()：如果 event.isSet()==False将阻塞线程；

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False

例子:
from threading import Thread,Event
import threading
import time,random

def conn_mysql():
    print(‘\033[42m%s 等待链接Mysql...\033[0m‘ %threading.current_thread().getName())
    event.wait()
    print(‘\033[42mMysql初始化成功，%s开始连接...\033[0m‘ %threading.current_thread().getName())

def check_mysql():
    print(‘\033[41m正在检查mysql...\033[0m‘)
    time.sleep(random.randint(1,3))
    event.set()
    time.sleep(random.randint(1,3))
if __name__ == ‘__main__‘:
    event = Event()
    t1 = Thread(target=conn_mysql)
    t2 = Thread(target=conn_mysql)
    t3 = Thread(target=check_mysql)

    t1.start()
    t2.start()
    t3.start()
    
wait(time)设置超时时间
from threading import Thread,Event
import threading
import time,random

def conn_mysql():
    while not event.is_set():
        print(‘\033[42m%s 等待连接mysql...\033[0m‘ %threading.current_thread().getName())
        event.wait(0.1)
    print(‘\033[42mMysql初始化成功，%s开始连接...\033[0m‘ %threading.current_thread().getName())

def check_mysql():
    print(‘\033[41m正在检查mysql...\033[0m‘)
    time.sleep(random.randint(1,3))
    event.set()
    time.sleep(random.randint(1,3))

if __name__ == ‘__main__‘:
    event=Event()
    t1 = Thread(target=conn_mysql)
    t2 = Thread(target=conn_mysql)
    t3 = Thread(target=check_mysql)
    t1.start()
    t2.start()
    t3.start()
    
Timer定时器，指定n秒后执行操作
例子:
from threading import Timer

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

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

四、协程
协程: 单线程下的并发，又称微线程，协程是一种用户态的轻量级线程，即协程是由用户程序自己控制
调度的
要实现协程，关键在于用户程序自己控制程序切换，切换之前必须由用户程序自己保存协程上一次调用
时的状态，如此，每次重新调用时，能够从上次的位置继续执行

我们之前已经学习过一种在单线程下可以保存程序运行状态的方法，即yield

不使用yield
import time

def consumer(item):
    x = 1111111111111
    y = 222222222222222
    z = 3333333333333333
    x1 = 122324234534534
    x2 = 21324354654654
    x3 = 3243565432435

def producer(target,seq):
    for item in seq:
        target(item)每次调用函数,会临时产生名称空间,调用结束则释放,循环100000000次,则重复这么多次的创建和释放,开销非常大
start_time = time.time()
producer(consumer,range(100000000))
stop_time = time.time()
print(‘run time is:%s‘ %(stop_time - start_time))
打印结果:run time is:14.8908851146698

使用yield
import time

def init(func):
    def wrapper(*args, **kwargs):
        g = func(*args, **kwargs)
        next(g)
        return g
    return wrapper

@init
def consumer():

    x = 1111111111111
    y = 222222222222222
    z = 3333333333333333
    x1 = 122324234534534
    x2 = 21324354654654
    x3 = 3243565432435
    while True:
        item = yield

def producer(target, seq):
    for item in seq:
        target.send(item)

start_time = time.time()
producer(consumer(), range(100000000))
stop_time=time.time()
print(‘run time is:%s‘ %(stop_time-start_time))

greenlet实现线程的切换
例子:
from greenlet import greenlet

def test1():
    print(‘test1,first‘)
    gr2.switch()
    print(‘test1,second‘)
    gr2.switch()

def test2():
    print(‘test2,first‘)
    gr1.switch()
    print(‘test2,second‘)
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

switch传参数
import time
from greenlet import greenlet
def eat(name):
    print(‘%s eat food 1‘ %name)
    gr2.switch(‘alex fly fly fly‘)
    print(‘%s eat food 2‘ %name)
    gr2.switch()

def play_phone(name):
    print(‘%s play 1‘ %name)
    gr1.switch()
    print(‘%s play 2‘ %name)

gr1 = greenlet(eat)
gr2=greenlet(play_phone)
gr1.switch(name=‘egon啦啦啦‘)

gevent第三方库
Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。
 
g1=gevent.spawn()创建一个协程对象g1
io阻塞切换
例子:
import gevent
import time

def eat():
    print(‘eat food 1‘)
    gevent.sleep(2)
    print(‘eat food 2‘)

def play_phone():
    print(‘play phone 1‘)
    gevent.sleep(1)
    print(‘play phone 2‘)

g1 = gevent.spawn(eat)
g2 = gevent.spawn(play_phone)
gevent.joinall([g1, g2])
print(‘主‘)

gevent.sleep(2)模拟的是gevent可以识别的io阻塞
time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码
例子:
from gevent import monkey;monkey.patch_all()
import gevent
import time

def eat():
    print(‘eat food 1‘)
    time.sleep(2)
    print(‘eat food 2‘)

def play_phone():
    print(‘play phone 1‘)
    time.sleep(1)
    print(‘play phone 2‘)

g1 = gevent.spawn(eat)
g2 = gevent.spawn(play_phone)
gevent.joinall([g1, g2])
print(‘主‘)


gevent实现单线程下的socket并发
例子:
服务端
from gevent import monkey;monkey.patch_all()
from socket import *
import gevent

def server(server_ip, port):
    s = socket(AF_INET, SOCK_STREAM)
    s.setsockopt(SOL_SOCKET,SO_REUSEADDR, 1)
    s.bind((server_ip,port))
    s.listen(5)

    while True:
        conn, addr = s.accept()
        gevent.spawn(talk, conn, addr)

def talk(conn,addr):
    try:
        while True:
            res = conn.recv(1024)
            print(‘client %s:%s msg: %s‘ %(addr[0], addr[1], res))
            conn.send(res.upper())
    except Exception as e:
        print(e)
    finally:
        conn.close()

if __name__ == ‘__main__‘:
    server(‘127.0.0.1‘, 8080)
    
客户端
#!/usr/bin/python
# --*-- coding: utf-8 --*--

from socket import *
client=socket(AF_INET, SOCK_STREAM)
client.connect((‘127.0.0.1‘, 8080))

while True:
    msg = input(‘>>: ‘).strip()
    if not msg:continue

    client.send(msg.encode(‘utf-8‘))
    msg = client.recv(1024)
    print(msg.decode(‘utf-8‘))

本文出自 “linux技术” 博客，请务必保留此出处http://haoyonghui.blog.51cto.com/4278020/1944191