python之线程和进程(并发编程)

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-22 下午8:45

# @Author  : LK

# @File    : lesson2.py

# @Software: PyCharm

import threading

import time

def music():

    print('begin listen music %s'%time.ctime())

    # t = input('请输入内容>>>')

    # time.sleep(3)

    print('stop listen music %s'%time.ctime())

    # print(t)

def game():

    print('begin to game %s'%time.ctime())

    time.sleep(5)

    print('stop to game %s'%time.ctime())

if __name__ == '__main__':

    t1 = threading.Thread(target=music)

    t2 = threading.Thread(target=game)

    t1.start()

    t2.start()

    # t1.join()   #   join就是等待的意思,让该线程执行完毕后,在执行主线程

    # t2.join()   # 注意如果注释这一句,结果是什么

    print('ending....')

    print(t1.getName())  #  获取线程名,

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-22 下午9:26

# @Author  : LK

# @File    : 守护线程.py

# @Software: PyCharm

#   守护线程 就是:和主线程一起退出,如果主线程结束了,那么不管守护的线程,有没有执行完毕,都退出

import threading

import time

def music():

    print('begin listen music %s' % time.ctime())

    time.sleep(3)

    print('stop listen music %s' % time.ctime())

def game():

    print('begin to game %s' % time.ctime())

    time.sleep(5)

    print('stop to game %s' % time.ctime())

t1 = threading.Thread(target=music)

t2 = threading.Thread(target=game)

threads = []

threads.append(t1)

threads.append(t2)

if __name__ == '__main__':

    # t1.setDaemon(True)

    t2.setDaemon(True)

    for t in threads:

        t.start()

        # t.setDaemon(True) # 守护线程, 就是和主线程一块结束

    print('ending....')

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-23 下午4:42

# @Author  : LK

# @File    : 同步锁.py

# @Software: PyCharm

import threading, time

#   用多线程去调用一个每次减1的函数,里面让他停留一会就会切换cpu,模拟线程安全问题,引出同步锁(互斥锁)

def sub():

    global num

    #  这样坐是0,因为num-=1 这个执行的很快,没有到达时间片切换就执行完了,每次取的都是不同的值

    # num -= 1

    # 如果这里让他停一会,就会发生,前面的一些拿到的值为100,然后开始停下来,让其他线程取值

    # 取得值还可能是100, 再过几个线程,取的时候就是前面的线程执行完了(到达时间片切换了)

    #  取得是就可能是其他值(前面几个线程执行后的结果),最终结果就不在是0

    #   这就是线程安全问题,多个线程都在处理同一个数据,

    # 如果处理的太慢(sleep)就会发生多个线程处理的值都是原始的值(多个线程处理一个值)

    #   处理方法: 就是在执行这三条语句时,不让cpu切换,知道处理完成时,再切换(同步锁)

    '''

    定义一个锁:lock = threading.Lock()

    lock.acquire()  #   锁起来(获取)

    这里是要执行的语句,

    lock.release()   #释放

    这里的不执行完,不切换cpu

    '''

    lock.acquire()

    temp = num

    time.sleep(0.001)

    num = temp - 1

    lock.release()

num = 100

l = []

lock = threading.Lock()

for i in range(100):

    t = threading.Thread(target=sub)

    l.append(t)

    t.start()

for t in l:

    t.join()

print(num)

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-23 下午8:33

# @Author  : LK

# @File    : 递归锁_死锁.py

# @Software: PyCharm

import threading, time

#   产生死锁:

'''

第一个线程执行doLockA, 然后lockB, 执行完actionA, 都释放了

开始执行actionB,对B上锁,不允许切换cpu,与此同时第二个线程开始执行actionA,对A上锁

不允许切换cpu,两个线程同时需要对方的资源释放但是都没有释放,所以死锁

(就像你问我要个香蕉, 我问你要个苹果, 你说我先给你,我说你先给我,于是就死锁了)

'''

class myThread(threading.Thread):

    def actionA(self):

        lockA.acquire()

        print(self.name, 'doLockA', time.ctime())

        time.sleep(3)

        lockB.acquire()

        print(self.name, 'doLockB', time.ctime())

        lockB.release()

        lockA.release()

    def actionB(self):

        lockB.acquire()

        print(self.name, 'doLockB', time.ctime())

        time.sleep(1)

        lockA.acquire()

        print(self.name, 'doLockA', time.ctime())

        lockA.release()

        lockB.release()

    def run(self):

        self.actionA()

        time.sleep(0.5)

        self.actionB()

if __name__ == '__main__':

    lockA = threading.Lock()

    lockB = threading.Lock()

    r_lock = threading.RLock()

    '''

    解决方法:

    用threading.RLock() 去替换锁A和锁B

    rlock就是在内部实现一个计数器,在同一个线程中,加锁就+1,释放就减一,只要锁数大于0,

    其他线程就不能申请锁,

    就是执行过程就是:当A函数执行完之后,所有线程开始抢占资源,谁抢到谁开始执行,

    比如线程2抢到, 线程2开始执行A函数,A没有执行完之前其他线程不能够继续执行

    '''

    threads = []

    for i in range(5):

        '''创建5个线程对象'''

        threads.append(myThread())

    for t in threads:

        t.start()

    for i in threads:

        t.join()

    print('ending.....')

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-24 下午4:45

# @Author  : LK

# @File    : 信号量.py

# @Software: PyCharm

'''

信号量, 也是一种锁的机制, 在那个递归锁中的同时事件的安全问题中

里面是多个线程同时抢占资源, 但是这个是只允许指定的个数,去抢占

'''

import threading, time

class myThread(threading.Thread):

    def run(self):

        if semaphore.acquire():

            print(self.name)

            time.sleep(3)

            semaphore.release()

if __name__ == '__main__':

    #   信号量, 参数的意思是同时允许几个线程去执行

    semaphore = threading.Semaphore(5)

    threads = []

    for i in range(10):

        threads.append(myThread())

    for t in threads:

        t.start()

    for t in threads:

        t.join()

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-23 下午9:38

# @Author  : LK

# @File    : 同步对象.py

# @Software: PyCharm

'''

event 同步对象标志位,就是在一个线程中设定后,在其他线程中也能捕获到

需求:一个boss类,一个worker类,

当boss对象执行后,worker才执行,就需要设置一个同步的标志位,用来判断

该执行那个线程

'''

'''

执行过程:

刚开始有5个worker线程和1个boss线程执行,

当worker执行时,wait没有被设定,一直在阻塞,但是boss可以执行

于是设定event并开始等5秒, 同时5个worker中的标志位变了,可以继续执行了

1秒后,worker清除event标志,继续阻塞,又过了4秒boss又设置了,

同时worker又可以执行了

'''

import threading, time

class Boss(threading.Thread):

    def run(self):

        print(self.name,'boss今天大家要加班到22点')

        print(event.isSet())  # false,判断是否标志

        event.set()

        time.sleep(5)

        print(self.name,'boss: <22:00>可以下班了')

        print(event.isSet())

        event.set()

class Worker(threading.Thread):

    def run(self):

        event.wait()  # 一旦event被设定 就等同于pass, 不设定就阻塞等待被设定

        print(self.name,'worker:哎呀生活苦呀,')

        time.sleep(1)

        event.clear()

        event.wait()

        print(self.name,'worker: oyear下班了')

#   如果不用event的话,就可能导致worker先说话,但是需求是让boss先说话,

#   因为cpu是抢占的方式执行的,不会按照需求走,所以就要引入同步标志位event

if __name__ == '__main__':

    event = threading.Event()

    threads =[]

    #   创建5个worker线程,1个boss线程

    for i in range(5):

        threads.append(Worker())

    threads.append(Boss())

    for t in threads:

        t.start()

    for t in threads:

        t.join()

    print('ending....')

'''

用多个线程同时对列表进行删除最后一个值,

会报错, 是因为可能会有多个线程同时取到最后一个值,都进行删除,

就会报错, 解决方法:可以用锁,将其锁住,(相当于串行的执行), 或者用队列

'''

'''

l = [1,2,3,4,5]

def fun():

    while 1:

        if not l:

            break

        a = l[-1]

        print(a)

        time.sleep(1)

        l.remove(a)

if __name__ == '__main__':

    t1 = threading.Thread(target=fun, args=())

    t2 = threading.Thread(target=fun, args=())

    t1.start()

    t2.start()

'''

import queue

#   创建一个队列,里面最多可以存放3个数据

q = queue.Queue(maxsize=3)  # FIFO,先进先出

# 将一个值入队

q.put(10)

q.put(9)

q.put(8)

#   如果这里在加上一个put, 就是有4个数据入队,但是空间不够,就会一直阻塞,直到,队列有空间时(从队列中取出get)

#   block参数默认是True, 如果改成False,当队列满的时候如果继续入队,就不会堵塞而是报错queue.Full

# q.put(9, block=False)

while 1:

    '''

    将一个值从队列中取出

    q.get()

    调用队列对象的get()

    方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，

    get()

    就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

    '''

    # 取完之后将会阻塞, block=Fales,如果队列为空时,继续出队,就会报错,queue.Empty

    #   get类似与put

    data = q.get(block=False)

    print(data)

# 先进后出  LIFO 类似与栈

import queue

# 先进后出  LIFO 类似与栈

q = queue.LifoQueue(maxsize=4)

q.put('你好')

q.put(123)

q.put({"name":"lucky"})

while 1:

    data = q.get()

    print(data)

# 还有在一种是优先级队列级别越低越先出来1级最低。class queue.PriorityQueue(maxsize)

import queue

q = queue.PriorityQueue(maxsize=3)

q.put([2,'你好'])

q.put([3,123,'aa'])

q.put([1,{"name":"lucky"}])

while 1:

    data = q.get()

    # print(data)

    print(data[1])  #   只显示数据,

import queue

q = queue.Queue(maxsize=4)

q.put('a')

q.put('b')

q.put('c')

print(q.qsize())  # 打印当前队列的大小,现在队列中有几个值

print(q.full())  # 判断是否满

print(q.empty())  # 判断是否为空

# q.put_nowait(3)  #  相当于q.put(3, block=Flase)

# q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号

# q.join() 实际上意味着等到队列为空，再执行别的操作

while 1:

    data = q.get()

    print(data)

    print(q.qsize())  # 打印当前队列的大小,现在队列中有几个值

    print(q.full())   # 判断是否满

    print(q.empty())  # 判断是否为空

from  multiprocessing import Process

import time

def f(name):

    time.sleep(1)

    print('hello  %s'%name, time.ctime())

if __name__ == '__main__':

    p_list = []

    for t in range(3):

        p = Process(target=f, args=('luck',))

        p_list.append(p)

    #   三个进程时在同一时刻开启的

    for p in p_list:

        p.start()

    for p in p_list:

        p.join()

    print('ending....')

class myProcess(Process):

    def __init__(self):

        # 父类初始化, 不传参数不用写

        super(myProcess, self).__init__()

    def run(self):

        time.sleep(1)

        print(self.name,'hello ',time.ctime())

if __name__ == '__main__':

    p_list = []

    for i in range(3):

        p_list.append(myProcess())

    for p in p_list:

        p.start()

    for p in p_list:

        p.join()

#   查查看进程的id

#   os.getppid()获得当前进程的父进程,os.getpid()获得当前进程

#   os.getppid 是pycharm的进程号

'''

from multiprocessing import Process

import os

import time

def info(title):

    print("title:", title)

    print('父进程 id:', os.getppid())

    print('当前进程 id:', os.getpid())

    # time.sleep(3)

if __name__ == '__main__':

    info('主进程')

    time.sleep(1)

    print("------------------")

    p = Process(target=info, args=('lucky子进程',))

    p.start()

    p.join()

    p_list = []

    for i in range(3):

        p = Process(target=info, args=('子进程%d' % i,))

        p_list.append(p)

    for p in p_list:

        p.start()

    for p in p_list:

        p.join()

Process 类

构造方法：

Process([group [, target [, name [, args [, kwargs]]]]])

　　group: 线程组，目前还没有实现，库引用中提示必须是None；

　　target: 要执行的方法；

　　name: 进程名；

　　args/kwargs: 要传入方法的参数。

实例方法：

　　is_alive()：返回进程是否在运行。

　　join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

　　start()：进程准备就绪，等待CPU调度

　　run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。

　　terminate()：不管任务是否完成，立即停止工作进程

属性：

　　daemon：和线程的setDeamon功能一样

　　name：进程名字。

'''

from multiprocessing import Process

import time

class myProcess(Process):

    def __init__(self, args):

        super(myProcess,self).__init__()

        self.args = args

    def fun(self):

        time.sleep(1)

        print(self.name, self.is_alive(), self.args, self.pid)

        time.sleep(1)

    def run(self):

        self.fun()

if __name__ == '__main__':

    p_list = []

    for i in range(10):

        # p = Process(target=fun, args=('子进程%d' % i,))

        p_list.append(myProcess('子进程%d' % i))

    # p_list[-2].daemon = True

    # 因为这三个进程是同时执行的,所以这个属性不明显

    for p in p_list:

        p.start()

    for p in p_list:

        p.join()

Managers支持的数据类型

# A manager returned by Manager() will support types list, dict, Namespace,

# Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier,

#  Queue, Value and Array. For example：

from multiprocessing import Manager

# Queue和pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。

# 所以利用Managers 进行数据共享, 同样也需要传参

def fun(d, l, i):

    d[i]='字典'+ str(i)

    l.append(i)

    # pass

if __name__ == '__main__':

    with Manager() as manager:

        d = manager.dict()  #  d = {} 创建一个字典

        l = manager.list(range(5)) # l = range(5)

        p_list = []

        for i in range(5):

            p = Process(target=fun,args=(d,l,i,))

            p.start()

            p_list.append(p)

        for p in p_list:

            p.join()

        print(d)

        print(l)

#   进程锁, lock

from multiprocessing import Process, Lock

import time

def fun(i):

    lock.acquire()

    print('%d'%i)

    lock.release()

if __name__ == '__main__':

    lock = Lock()

    for i in range(10):

        p = Process(target=fun, args=(i,)).start()

'''

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，

如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中有两个方法：

apply   同步

apply_async   异步(常用)

'''

from multiprocessing import Process, Pool

import time, os

def foo(i):

    time.sleep(1)

    print('子进程:',os.getpid(),i)

    return 'Hello %s'%i

def bar(args):

    # print('',os.getpgid())

    print(args)

    print('主进程:',os.getpid())

if __name__ == '__main__':

    pool = Pool(5)   #  进程池中开5个进程, 有100个任务需要执行, 每次执行5个

    print('main :',os.getpid())

    for i in range(100):

        #   回调函数:某个动作或者 函数执行成功后,再去执行的函数

        # pool.apply_async(func=foo, args=(i,))

        #   bar作为回调函数, 每次执行一个进程后都会执行回调函数, 回调函数是主进程中调用的

        #   func中函数的返回值,传给回调函数做参数

        pool.apply_async(func=foo, args=(i,), callback=bar)

    #   close和join必须加,而且顺序固定

    pool.close()

    pool.join()

from multiprocessing import Pool

import requests

import json

import os

def get_page(url):

    print('<进程%s> get %s' %(os.getpid(),url))

    respone=requests.get(url)

    if respone.status_code == 200:

        return {'url':url,'text':respone.text}

def pasrse_page(res):

    print('<进程%s> parse %s' %(os.getpid(),res['url']))

    parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))

    with open('db.txt','a') as f:

        f.write(parse_res)

if __name__ == '__main__':

    urls=[

        'https://www.baidu.com',

        'https://www.python.org',

        'https://www.openstack.org',

        'https://help.github.com/',

        'http://www.sina.com.cn/'

    ]

    p=Pool(3)

    res_l=[]

    for url in urls:

        res=p.apply_async(get_page,args=(url,),callback=pasrse_page)

        res_l.append(res)

    p.close()

    p.join()

    print([res.get() for res in res_l]) #拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# @Time    : 18-5-26 上午10:19

# @Author  : LK

# @File    : 协程.py

# @Software: PyCharm

from greenlet import greenlet

#   优缺点

'''

def test1():

    print(12)

    g2.switch()  #  切换到g2

    print(24)

    g2.switch()  #  切换到g2

def test2():

    print(56)

    g1.switch()  #  切换到g1

    print(65)

    g1.switch()

if __name__ ==  '__main__':

    g1 = greenlet(test1)

    g2 = greenlet(test2)

    g2.switch()

'''

'''

import gevent

import requests,time

start=time.time()

def f(url):

    print('GET: %s' % url)

    resp =requests.get(url)

    data = resp.text

    f = open('new.html', 'w', encoding='utf-8')

    f.write(data)

    f.close()

    print('%d bytes received from %s.' % (len(data), url))

#   这就相当于一个协程, 每次进行读写操作阻塞时,都会切换cpu

gevent.joinall([

        gevent.spawn(f, 'https://www.python.org/'),

        gevent.spawn(f, 'https://www.yahoo.com/'),

        gevent.spawn(f, 'https://www.baidu.com/'),

        gevent.spawn(f, 'https://www.sina.com.cn/'),

])

# f('https://www.python.org/')

#

# f('https://www.yahoo.com/')

#

# f('https://baidu.com/')

#

# f('https://www.sina.com.cn/')

print("cost time:",time.time()-start)

'''

# import threading, gevent

# def fun():

#     a = input('>>>')

#     print(a)

# def fun2():

#     print('\nhello')

# # gevent.joinall([

# #

# #         gevent.spawn(fun()),

# #         gevent.spawn(fun2()),

# # ])

# if __name__ == '__main__':

#     # t1 = threading.Thread(target=fun).start()

#     # t2 = threading.Thread(target=fun2).start()

#     gevent.joinall([

#

#             gevent.spawn(fun),

#             gevent.spawn(fun2),

#     ])