1、线程和进程python
#multiprocess.py import os print 'Process (%s) start....' % os.getpid() pid=os.fork() if pid==0: print 'I am child process (%s) and my parent is %s ' % (os.getpid(),os.getppid()) else: print 'i am father process'
form multiprocessing import Process import os #子进程要执行的代码 def run_proc(name): print 'Run child process %s (%s)' %(name,os.getpid()) if __name__=='__main__': print 'Parent process %s' % os.getpid() p=Process(target=run_proc,args=('test',)) print 'Process will start' p.start() p.join() print 'Process end'. 运行结果以下： Parent process 389 Process will start Run child process test (340) Process end
from multiprocessing import Pool import os,time,random def long_time_task(name): print 'Run task %s %s' %(name,os.getpid()) start = time.time() time.sleep(random.random*3) end=time.time() print 'Task %s runs %0.2f seconds' % (name,(end-start)) if __name__=='__main__' print 'Parent process %s' % os.getpid() p=Pool() for i in range(5): p.apply_async(long_time_task,args(i,)) print 'waiting for all subprocesses done.....' p.close() p.join() print 'All subprocesses done'
from multiprocessing import Process ,Queue import os,time,random #写数据进程执行的代码 def write(q): for value in ['A','B','C']: print 'Put %s to queue ....' %value q.put(value) time.sleep(random.random()) #读数据进程执行的代码: def read(q): while True: value=q.get(True) print 'Get %s from queue' %value if __name__=='__main__': #父进程建立Queue 并传给各个子进程 q=Queue() pw=Process(target=write,args=(q,)) pr=Process(target=read,args=(q,)) #启动子进程 pw pw.start() #启动子进程pr pr.start() #等待pw结束 pw.join() #pr进程里是死循环 没法等待结束 只能强行终止 pr.terminate()
#-*-coding:utf-8-*- import time,threading #新线程执行的代码 def loop(): print 'Thread %s is runing ..' % threading.current_thread().name n=0 while n<5: n=n+1 print 'thread %s >>> %s' % (threading.current_thread().name,n) time.sleep(1) print 'thread %s ended' % threading.current_thread().name print 'thread %s is runing ' % threading.current_thread().name t=threading.Thread(target=loop,name='LoopThread') t.start() t.join() print 'thread %s ended:' % threading.current_thread().name
#-*-coding:utf-8-*- import time,threading #假定这是银行存款 balance=0 lock=threading.Lock() def change_it(n): #先存后取 结果应该为0 global balance print '%s balance = %d' %(threading.current_thread().name,balance) balance=balance+n print '%s balance = %d' %(threading.current_thread().name,balance) balance=balance-n print '%s balance = %d' %(threading.current_thread().name,balance) def run_thread(n): for x in range(100000): lock.acquire() try: change_it(n) except Exception, e: raise e finally: lock.release() t1=threading.Thread(target=run_thread,args=(5,)) t2=threading.Thread(target=run_thread,args=(8,)) t1.start() t2.start() t1.join() t2.join() print balance
def process_student(name): std =Student(name) #std 是局部变量 可是每一个函数都要用它 所以必须穿进去 do_task_1(std) do_task_2(std) def do_task_1(std) do_subtask_1(std) do_subtask_2(std) def do_task_2(std) do_subtask_1(std) do_subtask_2(std)
#-*-coding:utf-8-*- import threading #建立全局ThreadLocal对象: local_school=threading.local() def process_student(): print 'Hello ,%s (in %s)' %(local_school.student,threading.current_thread().name) def process_thread(name): #绑定threadlocal的student local_school.student=name process_student() t1=threading.Thread(target=process_thread,args=('Alice',),name='Thread-A') t2=threading.Thread(target=process_thread,args=('Bob',),name='Thread-B') t1.start() t2.start() t1.join() t2.join()
#taskmanager.py import random,time,Queue from multiprocessing.managers import BaseManager #发送任务的队列 task_queue=Queue.Queue() #接受结果的队列 result_queue=Queue.Queue() #从basemanager 继承queuemanager class QueueManager(BaseManager): pass # 把两个Queue都注册到网上 callable参数关联了Queue对象： QueueManager.register('get_task_queue',callable=lambda:task_queue) QueueManager.register('get_result_queue,callable=lambda:result_queue') #绑定端口50000 设置验证码 abc manager=QueueManager(address=('',5000),authkey='abc') #启动Queue manager.start() #得到经过网络访问的Queue对象： task=manager.get_task_queue() result=manager.get_result_queue() #放几个任务进去 for i in range(10): n=random.randint(0,10000) print('Put task %d' %n) task.put(n) #从result队列读取结果 print('Try get results....'): for i in range(10): r=result.get(timeout=10) print('Result:%s' % r) #关闭 manager.shutdown()
在另外一台计算机上启动：web
#taskworker.py import time,sys,Queue from multiprocessing.managers import BaseManager #建立相似的QueueManager class QueueManager(BaseManager): pass #因为这个QueueManager 只从网上得到注册时只提供了名字 QueueManager.register('get_task_queue') QueueManager.register('get_task_result') #连接到服务器 也就是运行taskmanager.py的机器 server_addr='127.0.0.1' print('connect to server %s ..' % server_addr) #端口验证码一致 m=QueueManager(address=(server_addr,5000),authkey='abc') #从网络连接 m.connect() #获取queue对象 task=m.get_task_queue() result=m.get_task_result() #从task队列中取出数据 并把结果放到result队列中 for i in range(10): try: n=task.get(timeout=1) print('run task %d*%d' %(n,n)) r='%d*%d=%d' % (n,n,n*n) time.sleep(1) result.put(r) except Queue.Empty: print('task queue is empty') #处理结束 print('work exit')
这样就能够将任务拆分 并将任务发送到几台 几十台机器上进行处理。数据库
2、memcachewindows
理解一些概念：
Memcache是一个自由和开放源代码、高性能、分配的内存对象缓存系统。用于加速动态web应用程序，减轻数据库负载。它能够应对任意多个链接，使用非阻塞的网络IO。因为它的工做机制是在内存中开辟一块空间，而后创建一个HashTable，Memcached自管理这 些HashTable。Memcached是简单而强大的。它简单的设计促进迅速部署，易于发展所面临的问题，解决了不少大型数据缓存。它的API可供最流行的语言。Memcache是该系统的项目名称，Memcached是该系统的主程序文件，以守护程序方式运行于一个或多个服务器中，随时接受客 户端的链接操做，使用共享内存存取数据。
Memcached最吸引人的一个特性就是支持分布式部署；也就是说能够在一群机器上创建一堆Memcached服务，每一个服务能够根据具体服务器的硬件配置使用不一样大小的内存块，这样一来，理论上能够创建一个无限巨大的基于内存的cache storage系统。缓存
1.分别把memcached和libevent下载回来，放到 /tmp 目录下
Memcache用到了libevent这个库用于Socket的处理，因此还须要安装libevent。服务器
cd /tmp wget http://www.danga.com/memcached/dist/memcached-1.2.0.tar.gz wget http://www.monkey.org/~provos/libevent-1.2.tar.gz
先安装libevent网络
$ tar zxvf libevent-1.2.tar.gz $ cd libevent-1.2 $ ./configure –prefix=/usr $ make $ make install
3.测试libevent是否安装成功多线程
$ ls -al /usr/lib | grep libevent
2
lrwxrwxrwx 1 root root 21 11?? 12 17:38 libevent-1.2.so.1 -> libevent-1.2.so.1.0.3
3
-rwxr-xr-x 1 root root 263546 11?? 12 17:38 libevent-1.2.so.1.0.3
4
-rw-r–r– 1 root root 454156 11?? 12 17:38 libevent.a
5
-rwxr-xr-x 1 root root 811 11?? 12 17:38 libevent.la
6
lrwxrwxrwx 1 root root 21 11?? 12 17:38 libevent.so -> libevent-1.2.so.1.0.3并发
还不错，都安装上了。app
4.安装memcached，同时须要安装中指定libevent的安装位置
1
$ cd /tmp
2
$ tar zxvf memcached-1.2.0.tar.gz
3
$ cd memcached-1.2.0
4
$ ./configure –with-libevent=/usr ＃ 注意这个配置跟着libevent走
5
$ make
6
$ make install
若是中间出现报错，请仔细检查错误信息，按照错误信息来配置或者增长相应的库或者路径。
安装完成后会把memcached放到 /usr/local/bin/memcached ，
5.测试是否成功安装memcached
1
$ ls -al /usr/local/bin/mem*
2
-rwxr-xr-x 1 root root 137986 11?? 12 17:39 /usr/local/bin/memcached
3
-rwxr-xr-x 1 root root 140179 11?? 12 17:39 /usr/local/bin/memcached-debug
6 安装Python-memcached安装
memcached运行参数：
-d选项是启动一个守护进程，
-m是分配给Memcache使用的内存数量，单位是MB，我这里是10MB，
-u是运行Memcache的用户，我这里是root，
-l是监听的服务器IP地址，若是有多个地址的话，我这里指定了服务器的IP地址192.168.22.200(不指定为本机)
-p是设置Memcache监听的端口，我这里设置了12000，最好是1024以上的端口，
-c选项是最大运行的并发链接数，默认是1024，我这里设置了256，按照你服务器的负载量来设定，
-P是设置保存Memcache的pid文件，我这里是保存在 /tmp/memcached.pid，
python使用例子：
#!/usr/bin/env python import memcache mc = memcache.Client(['127.0.0.1:12000'],debug=0) mc.set("foo","bar") value = mc.get("foo") print value
Python-memcached API总结整个memcache.py只有1241行，至关精简主要方法以下：@set(key,val,time=0,min_compress_len=0)无条件键值对的设置，其中的time用于设置超时，单位是秒，而min_compress_len则用于设置zlib压缩(注:zlib是提供数据压缩用的函式库)@set_multi(mapping,time=0,key_prefix=”,min_compress_len=0)设置多个键值对，key_prefix是key的前缀，完整的键名是key_prefix+key, 使用方法以下>>> mc.set_multi({‘k1′ : 1, ‘k2′ : 2}, key_prefix=’pfx_’) == []>>> mc.get_multi(['k1', 'k2', 'nonexist'], key_prefix=’pfx_’) == {‘k1′ : 1, ‘k2′ : 2}@add(key,val,time=0,min_compress_len=0)添加一个键值对，内部调用_set()方法@replace(key,val,time=0,min_compress_len=0)替换value，内部调用_set()方法@get(key)根据key去获取value，出错返回None@get_multi(keys,key_prefix=”)获取多个key的值，返回的是字典。keys为key的列表@delete(key,time=0)删除某个key。time的单位为秒，用于确保在特定时间内的set和update操做会失败。若是返回非0则表明成功@incr(key,delta=1)自增变量加上delta，默认加1，使用以下>>> mc.set(“counter”, “20″)>>> mc.incr(“counter”)@decr(key,delta=1)自减变量减去delta，默认减1