Python网络编程之线程，进程

 

　　

　　一. 线程：

　　　　　　基本使用

　　　　　　线程锁

　　　　　　线程池

　　　　　　队列（生产者消费者模型）

　　二. 进程：

　　　　　  基本使用

　　　　　  进程锁

                进程池

                进程数据共享

   

    三. 协程：

　　　　　　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(
) : 唤醒一个或多个等待此条件变量的线程。

　　c.notify_all(): 唤醒所有等待此条件的线程。

import threading

def condition_func():

ret = False
inp = input('>>>')
if inp == '1':
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

View Code

　　

　　队列：

　　　　队列是线程间最常用的交换数据的形式。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

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

 

　很抱歉，时间有点仓促，写的不是很细，有点乱，以后慢慢补充整理，谢谢查看。