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([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 --10    -u root -l 127.0.0.1  -p 11511 -256 -/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

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

上一篇:[转]Google开源Leak Finder—用于检测内存泄漏的JavaScript工具-----可惜,暂时打不开google的网站,下载不了


下一篇:s32 kickstart 批量自动安装系统