2017年5月15日 – LeoKim's Blog

猴子补丁(monkey patch)

leokim / 2017年5月15日2017年5月15日 / Python

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

1	`import` `jsonprint` `'sub.py',json.__name__`

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

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

python 协程

leokim / 2017年5月15日 / Python

#生产者
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协作完成任务，所以称为“协程”，而非线程的抢占式多任务。

python 多线程

leokim / 2017年5月15日 / Python

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

python多进程

leokim / 2017年5月15日2017年5月15日 / Python

创建子进程

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

2017年5月
一	二	三	四	五	六	日
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30	31