python 多线程

一、参考出处

参考链接：https://docs.python.org/3/library/threading.html#module-threading

二、threading模块介绍

　　1）threading模块通过 Thread 创建线程

方式一：

#方式一
from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.start()
    print('主线程')

方式二：

#方式二
from threading import Thread
import time
class Sayhi(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        time.sleep(2)
        print('%s say hello' % self.name)


if __name__ == '__main__':
    t = Sayhi('egon')
    t.start()
    print('主线程')

　　2）threading/ Thread方法介绍：

Thread实例对象的方法
  isAlive(): 　　　　　　　　返回线程是否活动的。
  getName(): 　　　　　　　　返回线程名。
  setName(): 　　　　　　　　　　设置线程名。
　　
threading模块提供的一些方法：
  threading.currentThread(): 　　　　　　　　　　返回当前的线程变量。
  threading.enumerate(): 　　　　　　　　　　　　返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  threading.activeCount(): 　　　　　　　　　　返回正在运行的线程数量，与    len(threading.enumerate())有相同的结果。
　threading.RLock(*args, **kwargs):　　　　　　线程递归锁
　threading.Condition()　　　　　　　　　　　　　条件变量
　threading.Semaphore()
 hreading.BoundedSemaphore()
   threading.Event()　　　　　　　　　　

　　3）Thread  方法举例　

from threading import Thread
import threading
from multiprocessing import Process
import os

def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())


if __name__ == '__main__':
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()

    print(threading.current_thread().getName())
    print(threading.current_thread()) #主线程
    print(threading.enumerate()) #连同主线程在内有两个运行的线程
    print(threading.active_count())
    print('主线程/主进程')

    '''
    打印结果:
    MainThread
    <_MainThread(MainThread, started 140735268892672)>
    [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
    主线程/主进程
    Thread-1
    '''

　　4）threading模块介绍

　　RLock：

def RLock(*args, **kwargs):
    """Factory function that returns a new reentrant lock.

    A reentrant lock must be released by the thread that acquired it. Once a
    thread has acquired a reentrant lock, the same thread may acquire it again
    without blocking; the thread must release it once for each time it has
    acquired it.

    """
    if _CRLock is None:
        return _PyRLock(*args, **kwargs)
    return _CRLock(*args, **kwargs)

 样例：　同一线程中 可以多次申请锁。不同作用域中  应使用  同一个  RLock 创建的锁 多个  不同 RLock 锁 还是会造成死锁

from threading import Thread,Lock
import time
mutexB = mutexA=RLock()

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print('33[41m%s 拿到A锁33[0m' %self.name)

        mutexB.acquire()
        print('33[42m%s 拿到B锁33[0m' %self.name)
        mutexB.release()

        mutexA.release()

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

        mutexA.acquire()
        print('33[44m%s 拿到A锁33[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锁
然后就卡住，死锁了
'''

　Lock：

Lock = _allocate_lock
_allocate_lock = _thread.allocate_lock


def allocate_lock(): # real signature unknown; restored from __doc__
    """
    allocate_lock() -> lock object
    (allocate() is an obsolete synonym)
    
    Create a new lock object. See help(type(threading.Lock())) for
    information about locks.
    """
    pass

样例：  Lock 造成死锁。

from threading import Thread,Lock
import time

mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('%s 拿到了A锁' %self.name)

        mutexB.acquire()
        print('%s 拿到了B锁' %self.name)
        mutexB.release()

        mutexA.release()


    def f2(self):
        mutexB.acquire()
        print('%s 拿到了B锁' % self.name)
        time.sleep(0.1)

        mutexA.acquire()
        print('%s 拿到了A锁' % self.name)
        mutexA.release()

        mutexB.release()

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

Semaphore：　

def acquire(self, blocking=True, timeout=None):

def release(self):

 样例：

from threading import Thread,Semaphore,currentThread
import time,random

sm=Semaphore(3)

def task():
    # sm.acquire()
    # print('%s in' %currentThread().getName())
    # sm.release()
    with sm:
        print('%s in' %currentThread().getName())
        time.sleep(random.randint(1,3))


if __name__ == '__main__':
    for i in range(10):
        t=Thread(target=task)
        t.start()

Event：

 def is_set(self):
        """Return true if and only if the internal flag is true."""
        return self._flag

  def set(self):
        """Set the internal flag to true.”“”
    
      def clear(self):
        """Reset the internal flag to false.”“”

 def wait(self, timeout=None):
        """Block until the internal flag is true”“”

样例：

from threading import Thread,Event,currentThread
import time

event=Event()

def conn():
    n=0
    while not event.is_set():
        if n == 3:
            print('%s try too many times' %currentThread().getName())
            return
        print('%s try %s' %(currentThread().getName(),n))
        event.wait(0.5)
        n+=1

    print('%s is connected' %currentThread().getName())


def check():
    print('%s is checking' %currentThread().getName())
    time.sleep(5)
    event.set()


if __name__ == '__main__':
    for i in range(3):
        t=Thread(target=conn)
        t.start()
    t=Thread(target=check)
    t.start()

Condition：

class Condition:

    def wait(self, timeout=None):　　　　　调用这个方法将使线程进入Condition的等待池等待通知，并释放锁。使用前线程必须已获得锁定，否则将抛出异常。 　
        pass

    def wait_for(self, predicate, timeout=None):
        pass

    def notify(self, n=1)　　调用这个方法将从等待池挑选一个线程并通知，收到通知的线程将自动调用acquire()尝试获得锁定（进入锁定池）；其他线程仍然在等待池中。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。 
        pass

    def notify_all(self):　　 调用这个方法将通知等待池中所有的线程，这些线程都将进入锁定池尝试获得锁定。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
        pass

    def acquire(self, *args):　　　　调用关联的锁的相应方法
        pass

    def release(self):　　　　　　调用关联的锁的相应方法
        pass

 样例：

#coding=utf-8
__author__ = 'Bruce_Zhou'
import threading
import time
import datetime
num = 0
con = threading.Condition()

class Gov(threading.Thread):
    def __init__(self):
        super(Gov, self).__init__()

    def run(self):
        global num
        con.acquire()
        while True:
            print "开始拉升股市"
            num += 1
            print "拉升了" + str(num) + "个点"
            time.sleep(2)
            if num == 5:
                print "暂时安全！"
                con.notify()
                con.wait()
        con.release()


class Consumers(threading.Thread):
    def __init__(self):
        super(Consumers, self).__init__()

    def run(self):
        global num
        con.acquire()
        while True:
            if num > 0:
                print "开始打压股市"
                num -= 1
                print "打压了" + str(num) + "个点"
                time.sleep(2)
                if num == 0:
                    print "你妹的！天台在哪里！"
                    con.notify()
                    con.wait()
        con.release()

if __name__ == '__main__':
    p = Gov()
    c = Consumers()
    p.start()
    c.start()

 Timer

class Timer(Thread):
        def cancel(self):
        """Stop the timer if it hasn't finished yet."""
        self.finished.set()

    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()

 样例：

from threading import Timer
import random
import sys
class Code:
    def __init__(self):
        self.make_cache()

    def make_cache(self,interval=5):
        self.cache=self.make_code()
        print(self.cache)
        self.t=Timer(interval,self.make_cache)
        self.t.start()


    def make_code(self,n=4):
        res=''
        for i in range(n):
            s1=str(random.randint(0,9))
            s2=chr(random.randint(65,90))
            res+=random.choice([s1,s2])
        return res

    def check(self):
        while True:
            code=input('请输入你的验证码>>: ').strip()
            if code.upper() == self.cache:
                print('验证码输入正确')
                self.t.cancel()
                break


obj=Code()
obj.check()

concurrent.futures：进程线程池

　　ThreadPoolExecutor和ProcessPoolExecutor两个类，实现了对threading和multiprocessing的进一步抽象，对编写线程池/进程池提供了直接的支持。

参考：https://www.cnblogs.com/kangoroo/p/7628092.html

参考：https://www.cnblogs.com/dylan-wu/p/7163823.html

样例：

from concurrent.futures import ThreadPoolExecutor, Executor
import requests
import time

def get(url):
    print('GET %s' %url)
    response=requests.get(url)
    time.sleep(3)
    return {'url':url,'content':response.text}


def parse(res):
    res=res.result()
    print('%s parse res is %s' %(res['url'],len(res['content'])))


if __name__ == '__main__':
    urls=[
        'http://www.cnblogs.com/linhaifeng',
        'https://www.python.org',
        'https://www.openstack.org',
    ]

    pool=ThreadPoolExecutor(2)

    for url in urls:
        pool.submit(get,url).add_done_callback(parse)