Python之路_Day11

时间：2016-07-30 01:44:34 阅读：415 评论：0 收藏：0 [点我收藏+]

Python之路_Day11_课堂笔记

前期回顾

本节摘要

一、多线程

基本使用
生产者消费者模型(队列)
自定义线程池

二、多进程

基本使用
进程池

三、协程

greenlet
gevent

四、缓存

memcache
redis

~~五、rabbitMQ~~

六、下节预告：

MySQL
ORM框架-sqlchemy
堡垒机

一、线程

创建线程的两种基本使用方法：

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import threading
# def f1(arg):
#     print(arg)
#
# t = threading.Thread(target=f1,args=(123,))
# t.start()
class MyThread(threading.Thread):
    def __init__(self,func,args):
        self.func = func
        self.args = args
        super(MyThread,self).__init__()
    def run(self):
        self.func(self.args)
def f2(arg):
    print(arg)
obj = MyThread(f2,123)
obj.start()

先进先出队列Queue：

put放数据，是否阻塞，阻塞时的超时时间

get取数据（默认阻塞），是否阻塞，阻塞时的超时时间

队列最大长度

qsize()真实个数

maxsize最大支持的个数

join,task_done,阻塞进程，当队列中任务执行完毕后，释放阻塞，不再阻塞

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import queue
q = queue.Queue(3)
print(q.empty())
q.put(11)
q.put(22)
q.put(33)
print(q.empty())
print(q.qsize())
q.put(44)
q.put(55,block=False)
q.put(55,block=False,timeout=2)
print(q.get())
print(q.get())
print(q.get())
print(q.get(timeout=2))
q = queue.Queue(5)
q.put(123)
q.put(456)
q.get()
q.task_done()
q.get()
q.task_done()
q.join()

后进先出队列LifoQueue：

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import queue
q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())

优先级队列PriorityQueue：

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import queue
# 优先级队列
q = queue.PriorityQueue()
q.put((1,123))
q.put((2,456))
q.put((3,789))
print(q.get())

双向队列deque：

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import queue
# 双向队列
q = queue.deque()
q.append(123)
q.append(456)
q.appendleft(789)
q.pop()
q.popleft()

生产者消费者模型：

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
# 生产者消费者模型
import queue
import threading
import time
q = queue.Queue(20)
def productor(arg):
    ‘‘‘
    买票
    :param arg:
    :return:
    ‘‘‘
    q.put(str(arg) + ‘ 火车票 ‘)
for i in range(300):
    t = threading.Thread(target=productor,args=(i,))
    t.start()
def consumer(arg):
    ‘‘‘
    服务器后台
    :param arg:
    :return:
    ‘‘‘
    while True:
        print(arg,q.get())
        time.sleep(2)
for j in range(3):
    t = threading.Thread(target=consumer,args=(j,))
    t.start()

线程锁

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import threading
import time
NUM = 10
def func(l):
    global NUM
    # 上锁
    l.acquire()
    NUM -= 1
    time.sleep(2)
    print(NUM)
    # 开锁
    l.release()
lock = threading.RLock()    # 支持多层所，嵌套锁
# lock = threading.Lock()     # 只支持单层所
for i in range(10):
    t = threading.Thread(target=func , args=(lock,))
    t.start()

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import threading
def func(i,e):
    print(i)
    e.wait()    # 检测时什么灯，如果是红灯，停，绿灯，行；
    print(i+100)
event = threading.Event()
for i in range(10):
    t = threading.Thread(target=func,args=(i,event,))
    t.start()
event.clear()   # 设置成红灯
inp = input(‘>>>‘)
if inp == ‘1‘:
    event.set() # 设置成绿灯

线程池

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
import queue
import threading
import time
class ThreadPool:
    def __init__(self,maxsize=5):
        self.maxsize = maxsize
        self._q = queue.Queue(maxsize)
        for i in range(maxsize):
            self._q.put(threading.Thread)
    def get_thread(self):
        return self._q.get()
    def add_thread(self):
        self._q.put(threading.Thread)
pool = ThreadPool(5)
def task(arg, p):
    print(arg)
    time.sleep(1)
    p.add_thread()
for i in range(100):
    t = pool.get_thread()
    obj = t(target=task,args=(i,pool,))
    obj.start()

二、进程

基本使用

默认数据不共享

queues

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
# 进程基本使用，
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
def foo(i,arg):
    arg.put(i)
    print(‘say hi‘,i ,arg.qsize())
if __name__ == "__main__":
    li = queues.Queue(20,ctx=multiprocessing)
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()

array

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
# 进程基本使用2
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
from multiprocessing import Array
def foo(i,arg):
    arg[i] = i + 100
    for item in arg:
        print(item)
    print(‘============‘)
if __name__ == "__main__":
    li = Array(‘i‘,10)
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()

Manager.dict

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
# 进程基本使用3
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
from multiprocessing import Manager
def foo(i,arg):
    arg[i] = i + 100
    print(arg.values())
if __name__ == "__main__":
    obj = Manager()
    li = obj.dict()
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()
    import time
    time.sleep(1)

进程池

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
# 进程池
from multiprocessing import Pool
import time
def f1(arg):
    time.sleep(1)
    print(arg)
if __name__ == "__main__":
    pool = Pool(5)
    for i in range(30):
        # pool.apply(func=f1,args=(i,))
        pool.apply_async(func=f1,args=(i,))
    # pool.close()    # 所有的任务执行完毕
    # time.sleep(1)
    # pool.terminate()    # 立即终止
    # pool.join()

PS：

IO密集型使用多线程
计算密集型使用多进程

三、协程

原理：利用一个线程，分解一个线程成为多个“微线程” ===>程序级别

greenlet

gevent

#!/usr/bin/env python
# -.- coding: utf-8 -.-
# By sandler
from gevent import monkey ; monkey.patch_all()
import gevent
import requests
def f(url):
    print(‘GET: %s ‘ % url)
    resp = requests.get(url)
    data = resp.text
    print(‘%d bytes received from %s.‘ % (len(data),url ))
gevent.joinall([
    gevent.spawn(f,‘http://www.python.org‘),
    gevent.spawn(f,‘http://www.yahoo.com‘),
    gevent.spawn(f,‘http://www.github.com‘)
])