每日归档： 2017年5月15日

monkey patch指的是在运行时动态替换,一般是在startup的时候.

用过gevent就会知道,会在最开头的地方gevent.monkey.patch_all();把标准库中的thread/socket等给替换掉.这样我们在后面使用socket的时候可以跟平常一样使用,无需修改任何代码,但是它变成非阻塞的了.

之前做的一个游戏服务器,很多地方用的import json,后来发现ujson比自带json快了N倍,于是问题来了,难道几十个文件要一个个把import json改成import ujson as json吗?

其实只需要在进程startup的地方monkey patch就行了.是影响整个进程空间的.

同一进程空间中一个module只会被运行一次.

下面是代码.

main.py

import jsonimport ujsondef monkey_patch_json():
	json.__name__ = 'ujson'
	json.dumps = ujson.dumps
	json.loads = ujson.loads

monkey_patch_json()
print 'main.py',json.__name__
import sub

sub.py

import jsonprint 'sub.py',json.__name__

运行main.py,可以看到都是输出'ujson',说明后面import的json是被patch了的.

最后,注意不能单纯的json = ujson来替换.

#生产者
def consumer():
    r = ''
    while True:
        n = yield r
        if not n:
            return
        print('[CONSUMER] Consuming %s...' % n)
        r = '200 OK'


def produce(c):
	#启动生成器
    c.send(None)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] Producing %s...' % n)

        #切换到consumer执行
        r = c.send(n)

        print('[PRODUCER] Consumer return: %s' % r)
    c.close()

c = consumer()
produce(c)

注意到consumer函数是一个generator，把一个consumer传入produce后：

1. 首先调用c.send(None)启动生成器；

2. 然后，一旦生产了东西，通过c.send(n)切换到consumer执行；

3. consumer通过yield拿到消息，处理，又通过yield把结果传回；

4. produce拿到consumer处理的结果，继续生产下一条消息；

5. produce决定不生产了，通过c.close()关闭consumer，整个过程结束。

整个流程无锁，由一个线程执行，produce和consumer协作完成任务，所以称为“协程”，而非线程的抢占式多任务。

import time, threading

#新线程执行的代码：
def loop():
	print('thread %s is running...' % 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 running...' % threading.current_thread().name)
t1 = threading.Thread(target=loop, name='LoopThread1')
t1.start()
t2 = threading.Thread(target=loop, name='LoopThread2')
t2.start()
t3 = threading.Thread(target=loop, name='LoopThread3')
t3.start()
t4 = threading.Thread(target=loop, name='LoopThread4')
t4.start()

t1.join()
t2.join()
t3.join()
t4.join()
print('thread %s ended.' % threading.current_thread().name)

由于任何进程默认就会启动一个线程，我们把该线程称为主线程，主线程又可以启动新的线程，Python的threading模块有个current_thread()函数，它永远返回当前线程的实例。主线程实例的名字叫MainThread，子线程的名字在创建时指定，我们用LoopThread命名子线程。名字仅仅在打印时用来显示，完全没有其他意义，如果不起名字Python就自动给线程命名为Thread-1，Thread-2……

Lock

多线程和多进程最大的不同在于，多进程中，同一个变量，各自有一份拷贝存在于每个进程中，互不影响，而多线程中，所有变量都由所有线程共享，所以，任何一个变量都可以被任何一个线程修改，因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，把内容给改乱了。来看看多个线程同时操作一个变量怎么把内容给改乱了：

import time, threading

#假定这是你的银行存款：
balance = 0

def change_it(n):
	#先存后取，结果应该为0：
	global balance
	balance = balance + n
	balance = balance - n

def run_thread(n):
	for i in range(100000):
		change_it(n)

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)

结果跑出来会不一样，可以给线程加lock解决问题 因为这个现在还不怎么用到 所以就不深究了 可以在这里查看

http://www.liaoxuefeng.com/wiki/0014316089557264a6b348958f449949df42a6d3a2e542c000/00143192823818768cd506abbc94eb5916192364506fa5d000

创建子进程

from  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('Child process will start.')
 	p.start()
 	p.join()
 	print('Child process end.')

用Process来创建子进程

Process([group [, target [, name [, args [, kwargs]]]]]) 

target表示调用对象，args表示调用对象的位置参数元组。kwargs表示调用对象的字典。Name为别名。Group实质上不使用。

进程池

from multiprocessing import Pool
import os, time, random

def log_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())
	#pool默认大小是cpu核数
	p = Pool(4)
	for i in range(5):
		#pool.apply_async()用来向进程池提交目标请求
		p.apply_async(log_time_task, args=(i,))
	print('Wating for all subprocess done.')
	p.close()
	#pool.join()是用来等待进程池中的worker进程执行完毕,防止主进程在worker进程结束前结束.
	#但pool.join()必须使用在pool.close()或者pool.terminate()之后.
	#其中close()跟terminate()的区别在于close()会等待池中的worker进程执行结束再关闭pool,而terminate()则是直接关闭.
	p.join()
	print('All subprocesses done.')

	#result.successful()表示整个调用执行的状态，如果还有worker没有执行完，则会抛出AssertionError异常。
	#利用multiprocessing下的Pool可以很方便的同时自动处理几百或者上千个并行操作，脚本的复杂性也大大降低。

pool默认大小是cpu核数

pool.join()是用来等待进程池中的worker进程执行完毕,防止主进程在worker进程结束前结束. 

但pool.join()必须使用在pool.close()或者pool.terminate()之后. 

其中close()跟terminate()的区别在于close()会等待池中的worker进程执行结束再关闭pool,而terminate()则是直接关闭.

result.successful()表示整个调用执行的状态，如果还有worker没有执行完，则会抛出AssertionError异常。 

利用multiprocessing下的Pool可以很方便的同时自动处理几百或者上千个并行操作，脚本的复杂性也大大降低。

子进程

很多时候子进程并不是自身，而是一个外部进程。我们创建了子进程后，还需要进程的输入和输出。

subprocess模块可以让我们非常方便的启动一个子进程，然后控制其输入输出。

进程间通信

http://www.liaoxuefeng.com/wiki/0014316089557264a6b348958f449949df42a6d3a2e542c000/001431927781401bb47ccf187b24c3b955157bb12c5882d000