一. 线程：
基本使用
线程锁
线程池
队列（生产者消费者模型）
二. 进程：
基本使用
进程锁
进程池
进程数据共享
三. 协程：
gevent
greenlet
四. 缓存：
memcache
（一）线程：
所有的线程都运行于一个进程中，一个进程中可以执行多个线程。多个线程共享进程内的资源。所以可以将线程可以看成是共享同一虚拟内存以及其他属性的进程。
Threading用于提供线程相关的操作，线程是应用程序中工作的最小单元。
Thread(target=None, name=None, args=(), kwargs={}) : 创建一个新的线程实例
target：可调用对象，线程启动时，run()将调用此对象，默认为None
name: 线程名
args: 传递给target的参数元组
kwargs: 传递给target的关键字参数字典
Thread的实例： t.start: #线程准备就绪，等待CPU调度
t.run: #线程被cpu调度后自动执行线程对象的run方法
t.join([timeout]): #等待直到线程终止或者出现超时为止。
t.is_alive(): #如果线程是活动的。返回True，否则返回False
t.name: #线程名称
t.daemon: #线程的布尔型后台标志,必须在调用start()方法之前设置这个标志
###以线程的形式创建和启动一个函数：
import threading
import time def clock(interval):
while True:
print("The time is %s" % time.ctime())
time.sleep(interval) t = threading.Thread(target=clock,args=(5,))
#t.daemon = True
t.start() The time is Sat Jul 23 02:08:58 2016
The time is Sat Jul 23 02:09:03 2016
The time is Sat Jul 23 02:09:08 2016
###将同一个线程定义为一个类：
import threading
import time class ClockThread(threading.Thread):
def __init__(self,interval):
threading.Thread.__init__(self)
self.interval = interval
def run(self):
while True:
print("The time is %s" % time.ctime())
time.sleep(self.interval)
t = ClockThread(5)
t.start() The time is Sat Jul 23 02:15:48 2016
The time is Sat Jul 23 02:15:53 2016
The time is Sat Jul 23 02:15:58 2016
Timer对象：
定时器，在某个时间执行某个函数
格式：
Timer(interval, func [,args [, kwargs]])
对象：
p.start(): 启动定时器
p.cancel(): 如果函数尚未执行，取消定时器
from threading import Timer def hello():
print("hello, world") t = Timer(3, hello)
t.start() #3s后执行函数显示"hello，word" hello, world
信号量与有边界的信号量(semaphore)：
互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，每次调用acquire()方法时此计数器减1，每次调用release()方法时此计数器加1，如果计数器为0，acquire方法将会阻塞，直到其他线程调用release方法为止。比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。
Semaphore([value]) ：创建一个信号量，value为初始值，省略时，默认为1
p.acquire([blocking]):获取信号量
p.release() :通过将内部计数器值加1来释放一个信号量。
BoundedSemaphore([value]): 创建一个新的信号机，与Semaphore的工作方式完全相同，但是release()操作的次数不能超过acquire()操作次数
注： 信号机与互斥锁的差别在于：
信号机可用于发射信号，可以从不同线程调用以上两个方法。
import threading,time def run(n):
semaphore.acquire()
time.sleep(1)
print("run the thread: %s" %n)
semaphore.release() if __name__ == '__main__': num= 0
semaphore = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
for i in range(10):
t = threading.Thread(target=run,args=(i,))
t.start() run the thread: 0
run the thread: 4
run the thread: 3
run the thread: 2
run the thread: 1
run the thread: 5
run the thread: 9
run the thread: 8
run the thread: 7
run the thread: 6
事件(Event):
用于在线程之间通信。一个线程发出“事件”信号，一个或多个其它线程等待，Event实例管理者一个内部标志，可以使用set()方法将它置为True，clear()置为Flase, wait()方法将阻塞直到标志位True.
Event()
e.is_set(): 当内部标志位Ture时才返回True
e.set(): 将内部标志设置为True。等待它变为True的所有线程都将被唤醒
e.clear(): 将内部标志重置为False
e.wait([timeout]): 阻塞直到内部标志位True。
import threading def do(event):
print 'start'
event.wait()
print 'execute' event_obj = threading.Event()
for i in range(5):
t = threading.Thread(target=do, args=(event_obj,))
t.start() event_obj.clear()
inp = raw_input('input:')
if inp == 'true':
event_obj.set() start
start
start
start
start
input:true
execute
execute
execute
execute
execute
条件(Condition):
使得线程等待，只有满足某条件时，才释放n个线程
Condition([lock]) :创建一个条件变量,lock为可选的Lock或RLock实例，为指定则创建新的RLock实例供条件变量使用。
c.acquire(*args): 获取底层锁定
c.release(): 释放底层锁定
c.wait([timeout]): 等待直到获得通知或出现超时为止
c.notify([n]) : 唤醒一个或多个等待此条件变量的线程。
c.notify_all(): 唤醒所有等待此条件的线程。
import threading
def condition_func():
ret = False
inp = input('>>>')
if inp == '':
ret = True
return ret
def run(n):
con.acquire()
con.wait_for(condition_func)
print("run the thread: %s" %n)
con.release()
if __name__ == '__main__':
con = threading.Condition()
for i in range(10):
t = threading.Thread(target=run, args=(i,))
t.start()
>>>1
run the thread: 0
>>>1
run the thread: 1
>>>1
run the thread: 2
>>>1
run the thread: 3
线程池：
线程池是一个存放很多线程的单位，同时还有一个对应的任务队列。整个执行过程其实就是使用线程池中已有有限的线程把任务 队列中的任务做完。
import queue,threading,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(10):
#threading.Thread类
t = pool.get_thread()
obj = t(target = task, args = (i,pool))
obj.start() from threading import Timer
def hello():
print("hello,word")
t = Timer(3,hello)
t.start() 0
1
2
3
4
5
6
7
8
9
1
run the thread: 1
run the thread: 0
2
run the thread: 3
run the thread: 2
hello,word
3
run the thread: 4
4
5
6
7
8
9
10
队列：
队列是线程间最常用的交换数据的形式。queue模块是提供队列操作的模块,实现了各种多生产者，多使用者队列，可用于执行多个线程之间安全地交换信息。
queue模块定义了3种不同的队列类：
1. Queue([maxsize]): FIFO(先进先出)队列。maxsize为可放入项目的最大量。不设置或者为0时，队列无穷大。
2. LifoQueue([maxsize]): LIFO(后进先出)队列。也叫栈。
3. PriorityQueue([maxsize]): 优先级队列，项目按优先级从低到高排列，格式为(priority, data)形式的元组， priority为一个数字。
实例如下： 1 q.qsize(): #返回队列的正确大小
2 q.empty(): #如果队列为空，则返回True
3 q.full()：#如果队列已满，返回True
4 q.put(item [, block [, timeout): #将item放入队列. block，调用者将被阻塞直到队列中有可用的空闲位置即可。
5 q.put_nowait(item): #与q.put没什么差别
6 q.get([block [, timeout]]):3 从队列中删除一项，然后返回这个项目
7 q.get_nowait():#相当于get(0)
8 q.task_done(): 队列中数据的使用者用来指示对于项目的处理意见结束。
9 q.join(): 阻塞直到队列中的所有项目均被删除和处理为止。
案例：
（先进先出）
import queue
q = queue.Queue(2)
print(q.empty())
q.put(11)
q.put(22)
print(q.empty())
print(q.qsize())
print(q.get())
print(q.get())
q.put(33,block=False)
q.put(33,block=False,timeout=2)
print(q.get(timeout=2)) q = queue.Queue(5) q.put(123)
q.put(456)
print(q.get())
q.task_done()
print(q.get())
q.task_done()
q.join()
#queue.LifoQueue, #后进先出队列 q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())
# queue.PriorityQueue，优先级队列 # q = queue.PriorityQueue()
# q.put((8, 'hong'))
# q.put((2, 345))
# q.put((3, 678))
# print(q.get())
# queue.deque，双向对队 # q = queue.deque()
# q.append(123)
# q.append(333)
# q.appendleft(555)
#
# print(q.pop())
# print(q.popleft())
生产者与消费者模型：
生产者的工作是产生一块数据，放到buffer中，如此循环。与此同时，消费者在消耗这些数据（例如从buffer中把它们移除），每次一块。这里的关键词是“同时”。所以生产者和消费者是并发运行的，我们需要对生产者和消费者做线程分离。
import queue
import threading
import time
q = queue.Queue() def productor(arg):
"""
买票
:param arg:
:return:
"""
q.put(str(arg) + '- 买票') for i in range(20):
t = threading.Thread(target=productor,args=(i,))
t.start()
（二）进程：
进程是程序的一次执行，每个进程都有自己的地址空间，内存，数据栈。创建进程的时候，内核会为进程分配一定的资源，并在进程存活的时候不断进行调整，比如内存，进程创建的时候会占有一部分内存。进程结束的时候资源会释放出来，来让其他资源使用。我们可以把进程理解为一种容器，容器内的资源可多可少，但是只能进程间通信，不能共享信息。
谈到进程则要用到的就是multiprocessing模块，这个模块的所有功能基本都是在进程上的。
定义一个类运行一个进程：
process([,target [,name [,args [,kwargs]]]])
target: 当进程启动时执行的可调用对象
name: 为进程执行描述性名称的字符串
args: 位置参数，元组
kwargs: 位置参数，字典
通过这个构造函数简单构造了一个process进程。
进程（process）实例：
p.is_alive() #如果p仍然运行，返回True
p.join([timeout]) #等待进程p终止，timeout是可选的超时期限。进程可被连接无数次，但连接自身时则会报错
p.run()# 启动进程时运行的方法，可调用target。
p.start() #启动进程，代表进程的子进程，并调用p.run()函数
p.terminate()#强制终止进程。进程p被立即终止，而不会进行清理，慎用。
单进程实例：
import multiprocessing
import time def clock(interval):
while True:
print("The time is %s" % time.ctime())
time.sleep(interval)
if __name__ == '__main__':
p = multiprocessing.Process(target=clock, args=(5,))
p.start() The time is Fri Jul 22 17:15:45 2016
The time is Fri Jul 22 17:15:50 2016
The time is Fri Jul 22 17:15:55 2016
将上面的进程定义为继承自Process的类，目的为为了实现跨平台的可移植性，必须有主程序创建进程。
import multiprocessing
import time class ClockProcess(multiprocessing.Process):
def __init__(self, interval):
multiprocessing.Process.__init__(self)
self.interval = interval def run(self):
while True:
print("The time is %s" % time.ctime())
time.sleep(self.interval)
if __name__ == '__main__':
p = ClockProcess(5)
p.start() The time is Fri Jul 22 17:25:08 2016
The time is Fri Jul 22 17:25:13 2016
The time is Fri Jul 22 17:25:18 2016
进程锁：
当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。
import multiprocessing
import sys def worker_with(lock, f):
with lock:
fs = open(f,"a+")
fs.write('Lock acquired via \n')
fs.close() def worker_no_with(lock, f):
lock.acquire()
try:
fs = open(f,"a+")
fs.write('Lock acquired directly\n')
fs.close()
finally:
lock.release() if __name__ == "__main__": f = "file.txt" lock = multiprocessing.Lock()
w = multiprocessing.Process(target=worker_with, args=(lock, f))
nw = multiprocessing.Process(target=worker_no_with, args=(lock, f)) w.start()
nw.start() w.join()
nw.join() #cat file.txt Lock acquired directly
Lock acquired via
注：如果两个进程没有使用lock来同步,则他们对同一个文件的写操作可能会出现混乱。
进程池：
进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。
创建一个进程池：
Pool([numprocess [,initializer [, initargs]]])
numprocess: 要创建的进程数
initlalizer: 每个工作进程启动时要执行的可调用对象，默认为None
initargs： 传递给initlalizer的元组
Pool的实例：
p.apply(func, [, args [, kwargs]])#在一个池工作进程中执行函数(**args, **kwargs),然后返回结果，不能再池中并行执行，可使用apply_async
p.apply_async(func, [, args [, kwargs [,callback]]])#在一个池工作进程中异步执行函数(**args, **kwargs),然后返回结果，传递给callback。
p.terminate()#立即终止
p.close()# 关闭进程池
p.join()# 等待所有工作进程退出
案例：
from multiprocessing import Pool
import time def f1(arg):
time.sleep(3)
print(arg) if __name__ == '__main__':
pool = Pool(5) #并发执行5个函数 for i in range(15):
#pool.apply(func=f1,args=(i,))#不能并发的执行函数
pool.apply_async(func=f1,args=(i,))#可并发执行函数 pool.close() #所有的任务执行完毕
time.sleep(3)
#pool.terminate()#立即终止
pool.join()
进程数据共享：
通常进程之间是完全孤立的，使用数据共享，可以访问多个进程。
实现进程数据共享有两种方法：
#方法一，Array from multiprocessing import Process
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()
#方法二：manage.dict()共享数据 from multiprocessing import Process
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)
线程锁(Lock, RLock):
由于线程之间是进行随机调度，并且每个线程可能只执行n条执行之后，当多个线程同时修改同一条数据时可能会出现脏数据，所以，出现了线程锁 - 同一时刻允许一个线程执行操作。
Lock()：创建新的Lock对象，初始化状态为非锁定
lock.acquire([blocking]): 获取锁定
lock.release(): 释放锁定
import threading,time def run(n):
semaphore.acquire()
time.sleep(1)
print("run the thread: %s" %n)
semaphore.release() if __name__ == '__main__': num= 0
semaphore = threading.BoundedSemaphore(2) #最多允许5个线程同时运行
for i in range(5):
t = threading.Thread(target=run,args=(i,))
t.start() 1
run the thread: 1
run the thread: 0
2
run the thread: 3
run the thread: 2
3
run the thread: 4
4
5
6
7
8
9
10
（三）协程：
协程我们可以看成是一种用户空间的线程，利用一个线程，分解一个线程成为多个“微线程”
Python通过yield提供了对协程的基本支持，但是不完全。而第三方的gevent为Python提供了比较完善的协程支持。
gevent是第三方库，通过greenlet实现协程，其基本思想是：
当一个greenlet遇到IO操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO。
由于切换是在IO操作时自动完成，所以gevent需要修改Python自带的一些标准库，这一过程在启动时通过monkey patch完成：
from gevent import monkey; monkey.patch_socket()
import gevent def f(n):
for i in range(n):
print gevent.getcurrent(), i g1 = gevent.spawn(f, 5)
g2 = gevent.spawn(f, 5)
g3 = gevent.spawn(f, 5)
g1.join()
g2.join()
g3.join() <Greenlet at 0x10e49f550: f(5)> 0
<Greenlet at 0x10e49f550: f(5)> 1
<Greenlet at 0x10e49f550: f(5)> 2
<Greenlet at 0x10e49f550: f(5)> 3
<Greenlet at 0x10e49f550: f(5)> 4
<Greenlet at 0x10e49f910: f(5)> 0
<Greenlet at 0x10e49f910: f(5)> 1
<Greenlet at 0x10e49f910: f(5)> 2
<Greenlet at 0x10e49f910: f(5)> 3
<Greenlet at 0x10e49f910: f(5)> 4
<Greenlet at 0x10e49f4b0: f(5)> 0
<Greenlet at 0x10e49f4b0: f(5)> 1
<Greenlet at 0x10e49f4b0: f(5)> 2
<Greenlet at 0x10e49f4b0: f(5)> 3
<Greenlet at 0x10e49f4b0: f(5)> 4
可以看到，3个greenlet是依次运行而不是交替运行。
要让greenlet交替运行，可以通过gevent.sleep()交出控制权：
def f(n):
for i in range(n):
print gevent.getcurrent(), i
gevent.sleep(0) <Greenlet at 0x10cd58550: f(5)> 0
<Greenlet at 0x10cd58910: f(5)> 0
<Greenlet at 0x10cd584b0: f(5)> 0
<Greenlet at 0x10cd58550: f(5)> 1
<Greenlet at 0x10cd584b0: f(5)> 1
<Greenlet at 0x10cd58910: f(5)> 1
<Greenlet at 0x10cd58550: f(5)> 2
<Greenlet at 0x10cd58910: f(5)> 2
<Greenlet at 0x10cd584b0: f(5)> 2
<Greenlet at 0x10cd58550: f(5)> 3
<Greenlet at 0x10cd584b0: f(5)> 3
<Greenlet at 0x10cd58910: f(5)> 3
<Greenlet at 0x10cd58550: f(5)> 4
<Greenlet at 0x10cd58910: f(5)> 4
<Greenlet at 0x10cd584b0: f(5)> 4
3个greenlet交替运行，
把循环次数改为500000，让它们的运行时间长一点，然后在操作系统的进程管理器中看，线程数只有1个。
当然，实际代码里，我们不会用gevent.sleep()去切换协程，而是在执行到IO操作时，gevent自动切换，代码如下：
from gevent import monkey; monkey.patch_all()
import gevent
import urllib2 def f(url):
print('GET: %s' % url)
resp = urllib2.urlopen(url)
data = resp.read()
print('%d bytes received from %s.' % (len(data), url)) gevent.joinall([
gevent.spawn(f, 'https://www.python.org/'),
gevent.spawn(f, 'https://www.yahoo.com/'),
gevent.spawn(f, 'https://github.com/'),
]) GET: https://www.python.org/
GET: https://www.yahoo.com/
GET: https://github.com/
45661 bytes received from https://www.python.org/.
14823 bytes received from https://github.com/.
304034 bytes received from https://www.yahoo.com/.
从结果看，3个网络操作是并发执行的，而且结束顺序不同，但只有一个线程。
（四）缓存
memcache:
下载： wget http://ftp.tummy.com/pub/python-memcached/old-releases/python-memcached-1.54.tar.gz(自己更新最新版)
解压缩：tar -zxvf python-memcached-1.54.tar.gz
安装： python setup.py install
启动：memcached -d -m 10 -u root -l 127.0.0.1 -p 11511 -c 256 -P /tmp/memcached.pid
参数说明: -d 是启动一个守护进程
-m 是分配给Memcache使用的内存数量，单位是MB
-u 是运行Memcache的用户
-l 是监听的服务器IP地址
-p 是设置Memcache监听的端口,最好是1024以上的端口
-c 选项是最大运行的并发连接数，默认是1024，按照你服务器的负载量来设定
-P 是设置保存Memcache的pid文件
代码：
import memcache class MemcachedClient():
''' python memcached 客户端操作示例 ''' def __init__(self, hostList):
self.__mc = memcache.Client(hostList); def set(self, key, value):
result = self.__mc.set("name", "hongfei")
return result def get(self, key):
name = self.__mc.get("name")
return name def delete(self, key):
result = self.__mc.delete("name")
return result if __name__ == '__main__':
mc = MemcachedClient(["127.0.0.1:11511", "127.0.0.1:11512"])
key = "name"
result = mc.set(key, "hongfei")
print("set的结果：", result)
name = mc.get(key)
print ("get的结果：", name)
result = mc.delete(key)
print ("delete的结果：", result) set的结果： True
get的结果： hongfei
delete的结果： 1
很抱歉，时间有点仓促，写的不是很细，有点乱，以后慢慢补充整理，谢谢查看。