python3 进程和线程（一）

进程和线程

1. 进程：内存独立，线程共享同一进程的内存，一个进程就像一个应用程序，例如qq和word，这属于两个进程，
2. 进程是资源的集合，线程是执行单位
3. 进程之间不能直接互相访问，同一进程中的程可以互相通信
4. 创建新进程消耗系统资源，线程非常轻量，只保存线程运行时的必要数据，如上下文、程序堆栈信息
5. 同一进程里的线程可以相互控制，父进程可以控制子进程

import threading
import time

def sayhi(num):
    print('num:',num)
    time.sleep(3)

a = threading.Thread(target=sayhi,args=(1,))
b = threading.Thread(target=sayhi,args=(2,))
now1 = time.time()
print(now1)
a.start()
b.start()
now2 = time.time()
print(now2)
print(threading.active_count()) 
# 包含主线程，总共3个
print(a.getName())
print(b.getName())

class MyThread(threading.Thread):
    def __init__(self,n):
        threading.Thread.__init__(self)
        self.n = n

        def run(self):
            print('running on thread $s'%self.n)
            time.sleep(3)
now3 = time.time()
print(now3)
t1 = MyThread(1)
t2 = MyThread(2)
t1.start()
t2.start()
print(t1.getName())
print(t2.getName())
now4 = time.time()
print(now4)

thread_list = []
for i in range(10):
    s1 = threading.Thread(target=sayhi,args=(i,))
    s1.start()
    thread_list.append(s1)
now5 = time.time()
print(now5)
for r in thread_list:
    r.join() # s1.wait()
print('--work done--')
now6 = time.time()
print(now6)
print('primary'.center(20,'-'))


for ii in range(10):
    s2 = threading.Thread(target=sayhi)
    s2.setDaemon(s2)
    s2.start()

一、线程的基本使用 threading模块

threading 模块建立在 _thread 模块之上。

thread 模块以低级、原始的方式来处理和控制线程，

而 threading 模块通过对 thread 进行二次封装，提供了更方便的 api 来处理线程。

import threading
import time

def task(arg):
    time.sleep(arg)
    print(arg)

for i in range(30):
    # 创建线程，args必须是可迭代的对象
    t = threading.Thread(target=task,args=[i,])

    # 主线程终止，不等待子线程
    # t.setDaemon(True)
    # 等待子线程
    # t.setDaemon(False)

    # 开始工作,等待cpu使用
    t.start()

    # 变成串行
    # t.join() #一直等
    # t.join(1) # 超时，等待最大时间
print('end')

使用线程中的run方法

import threading
import time
class MyThread(threading.Thread): #继承threading方法
    def __init__(self,func,*args,**kwargs):
        super(MyThread,self).__init__(*args,**kwargs)
        self.func = func

    def run(self): # 线程中的run方法
        self.func()

def task():
    time.sleep(1)
    print('is a test')

obj = MyThread(func=task)
obj.start()

Thread方法说明

t.start() : 激活线程，

t.getName() : 获取线程的名称

t.setName() ： 设置线程的名称 

t.name : 获取或设置线程的名称

t.is_alive() ： 判断线程是否为激活状态

t.isAlive() ：判断线程是否为激活状态

t.setDaemon() 设置为后台线程或前台线程（默认：False）;通过一个布尔值设置线程是否为守护线程，必须在执行start()方法之后才可以使用。如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止；如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止

t.isDaemon() ： 判断是否为守护线程

t.ident ：获取线程的标识符。线程标识符是一个非零整数，只有在调用了start()方法之后该属性才有效，否则它只返回None。

t.join() ：逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义

t.run() ：线程被cpu调度后自动执行线程对象的run方法

二、线程锁threading.RLock和threading.Lock

由于线程之间是进行随机调度，并且每个线程可能只执行n条执行之后，CPU接着执行其他线程。为了保证数据的准确性，引入了锁的概念

假如现在有10个人要上厕所，钥匙在我手里，这个时候A进来，他把门关上了，别人就进不去，只有等A完事后其他人才能进来，这个门就是控制线程的那把锁

锁的基本使用：

lock=threading.Lock() 
lock.acquire() 
lock.release()

Rlock和Lock的区别：

　　RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。 如果使用RLock，那么acquire和release必须成对出现，即调用了n次acquire，必须调用n次的release才能真正释放所占用的琐

# 只能一个人使用锁
# 创建锁
lock = threading.Lock()
# 递归锁,多把锁
# lock = threading.RLock()

def task(arg):
    time.sleep(2)
    print('arg:',arg)
    # 申请使用锁，其他人等
    lock.acquire()
    # lock.acquire() # Rlock可使用多把锁

    global v
    v -= 1
    print(v)

    # 释放锁
    lock.release()
    # lock.release() # Rlock可解锁

for i in range(2):
    t = threading.Thread(target=task,args=(i,))
    t.start()

如果厕所有3个坑，同时允许3个人上，使用 BoundedSemaphore

# 多个人同时使用锁
# 信号链，定义3个人同时使用
lock = threading.BoundedSemaphore(3)
def task(arg):
    lock.acquire()
    time.sleep(1)
    global v
    v -= 1
    print(v)
    lock.release()
for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()

三、事件Event

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

• clear：将“Flag”设置为False
• set：将“Flag”设置为True
• Event.isSet() ：判断标识位是否为Ture

# 事件 event
lock = threading.Event()
def task(arg):
    time.sleep(1)
    # 锁住所有的线程
    lock.wait()
    print(arg)
for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()
while 1:
    value = input('>>:').strip()
    if value == '1':
        lock.set() # 打开锁，执行上面的print
        # lock.clear() # 再锁上

四、条件Condition

condition变量服从上下文管理协议：with语句块封闭之前可以获取与锁的联系。

acquire() 和 release() 会调用与锁相关联的相应的方法。

其他和锁关联的方法必须被调用，wait()方法会释放锁，当另外一个线程使用 notify() or notify_all()唤醒它之前会一直阻塞。一旦被唤醒，wait()会重新获得锁并返回，

• wait(timeout=None) ： 等待通知，或者等到设定的超时时间。当调用这wait()方法时，如果调用它的线程没有得到锁，那么会抛出一个RuntimeError 异常。 wati()释放锁以后，在被调用相同条件的另一个进程用notify() or notify_all() 叫醒之前 会一直阻塞。wait() 还可以指定一个超时时间。

如果有等待的线程，notify()方法会唤醒一个在等待conditon变量的线程。notify_all() 则会唤醒所有在等待conditon变量的线程。

lock = threading.Condition()

def task(arg):
    time.sleep(1)
    # 锁住所有线程
    lock.acquire()
    lock.wait()
    print('线程：',arg)
    lock.release()
for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()
while 1:
    value = input('>>>:').strip()
    lock.acquire()
    lock.notify(int(value))
    lock.release()

>>>:1
线程： 3
>>>:2
线程： 4
线程： 2
>>>:3
线程： 0
线程： 7
线程： 1
>>>:4
线程： 8
线程： 9
线程： 6
线程： 5
>>>:5

五、queue模块

Queue 就是对队列，它是线程安全的

举例来说，我们去麦当劳吃饭。饭店里面有厨师职位，前台负责把厨房做好的饭卖给顾客，顾客则去前台领取做好的饭。这里的前台就相当于我们的队列。形成管道样，厨师做好饭通过前台传送给顾客，所谓单向队列

import queue
 
q = queue.Queue(maxsize=0)  # 构造一个先进显出队列，maxsize指定队列长度，为0 时，表示队列长度无限制。
 
q.join()    # 等到队列为kong的时候，在执行别的操作
q.qsize()   # 返回队列的大小 （不可靠）
q.empty()   # 当队列为空的时候，返回True 否则返回False （不可靠）
q.full()    # 当队列满的时候，返回True，否则返回False （不可靠）
q.put(item, block=True, timeout=None) #  将item放入Queue尾部，item必须存在，可以参数block默认为True,表示当队列满时，会等待队列给出可用位置，
　　　　　　　　　　　　　　　　　　　　　　　　 为False时为非阻塞，此时如果队列已满，会引发queue.Full 异常。 可选参数timeout，表示 会阻塞设置的时间，过后，
　　　　　　　　　　　　　　　　　　　　　　　　  如果队列无法给出放入item的位置，则引发 queue.Full 异常
q.get(block=True, timeout=None) #   移除并返回队列头部的一个值，可选参数block默认为True，表示获取值的时候，如果队列为空，则阻塞，为False时，不阻塞，
　　　　　　　　　　　　　　　　　　　　　　若此时队列为空，则引发 queue.Empty异常。 可选参数timeout，表示会阻塞设置的时候，过后，如果队列为空，则引发Empty异常。
q.put_nowait(item) #   等效于 put(item,block=False)
q.get_nowait() #    等效于 get(item,block=False)

六、线程池

from concurrent.futures import ThreadPoolExecutor as TPE

def task(arg):
    time.sleep(0.5)
    print('Thread:',arg)

pool = TPE(5) # 线程池里放5个线程

for i in range(100):
    # 去连接池中获取连接
    pool.submit(task,i)

import requests
def task(url):
    response = requests.get(url)
    print('得到结果：',url,len(response.content))

pool = TPE(2)
url_list = [
    'http://www.oldboyedu.com',
    'http://www.baidu.com',
    'http://www.sohu.com',
]

for url in url_list:
    print('开始请求',url)
    # 去连接池获取连接
    pool.submit(task,url)

七、回调函数

线程池定义为一个变量，使用变量.add_done_callback(函数名称)进行函数回调

import requests

def txt(future):
    download_response = future.result()
    print('得到结果：', url, len(download_response.content))

def download(url):
    response = requests.get(url)
    # print('得到结果：',url,len(response.content))
    return response
pool = TPE(2)
url_list = [
    'http://www.oldboyedu.com',
    'http://www.baidu.com',
    'http://www.sohu.com',
]

for url in url_list:
    print('开始请求',url)
    # 去连接池获取连接
    future = pool.submit(download,url)
    # 一旦download函数return，开始执行txt函数
    future.add_done_callback(txt)

八、进程

multiprocessing是python的多进程管理包，和threading.Thread类似。

from multiprocessing import Process
def task(arg):
    time.sleep(arg)
    print(arg)

if __name__ == '__main__':
    for i in range(10):
        p = Process(target=task,args=(i,))
        # 守护true，不执行子进程
        # p.daemon = True
        # false执行子进程，默认
        # p.daemon = False
        p.start()
        # p.join()
        p.join(1)
    print('主进程中的主线程...')

九、multiprocessing，Array,Value

数据可以用Value或Array存储在一个共享内存地图里，如下：

# 进程数据共享
from multiprocessing import Process,Array
def task(num,li):
    li[num] = num
    print(list(li))

if __name__ == '__main__':
    L = Array('i',10) # 数据类型，长度
    for i in range(10):
        p = Process(target=task,args=(i,L))
        p.start()

from multiprocessing import Array, Value, Process


def func(a, b):
    a.value = 6.66666
    for i in range(len(b)):
        b[i] = -b[i]


if __name__ == "__main__":
    num = Value('d', 0.0)
    arr = Array('i', range(11))

    c = Process(target=func, args=(num, arr))
    d = Process(target=func, args=(num, arr))
    c.start()
    d.start()
    c.join()
    d.join()

    print(num.value)
    for i in arr:
        print(i)

十、进程池

和线程池差不多

from concurrent.futures import ProcessPoolExecutor as PPE


#基本用法
def task(arg):
    time.sleep(1)
    print(arg)

pool = PPE(5)
for i in range(10):
    pool.submit(task,i)


# 进程池回调
def call(arg):
    data = arg.result()
    print(data)

def task(arg):
    print(arg)
    return arg+100

pool = PPE(5)
for i in range(10):
    obj = pool.submit(task,i)
    obj.add_done_callback(call)