sample
Python Multiprocessing Example How To Use Signal Driven Programming In Applications
闭包、装饰器、生成器、协程、多进程
mixin类似java中的implement, PHP中的trait, 定义的是接口，而非类
协程的特点在于是一个线程执行，那和多线程比，协程有何优势？
最大的优势就是协程极高的执行效率。因为子程序切换不是线程切换，而是由程序自身控制，因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显。
第二大优势就是不需要多线程的锁机制，因为只有一个线程，也不存在同时写变量冲突，在协程中控制共享资源不加锁，只需要判断状态就好了，所以执行效率比多线程高很多。
因为协程是一个线程执行，那怎么利用多核CPU呢？
最简单的方法是多进程+协程，既充分利用多核，又充分发挥协程的高效率，可获得极高的性能。
===闭包===
def countdown(n):
def next():
nonlocal n # 声明n为next()函数外部的变量
r = n
n -= 1
return r
return next
next = countdown(10)
while True:
v = next()
if not v: break
===装饰器===
# 定义装饰器函数
def print_result(func):
def callf(*args, **kwargs):
r = func(*args, **kwargs)
print('Do something with return: %d.' %r)
return r+7
return callf
# 用@装饰原函数
@print_result
def square(x):
return x*x
# 多个装饰器
@dec1
@dec2
@dec3
def square(x):
pass
square = dec1(dec2(dec3(square)))
# 装饰器参数
@eventhandler('BUTTON')
def handle_button(msg):
pass
@eventhandler('RESET')
def handle_reset(msg):
pass
# 事件处理程序装饰器
event_handlers = {}
def eventhandler(event):
def register_function(f):
event_handlers[event] = f
return f
return register_function
@eventhandler('BUTTON')
def handle_button(msg):
pass
===生成器===
# 定义生成器
def countdown(n):
print('Start counting')
while n > 0:
yield n
n -= 1
for i in countdown(3):
print(i)
# 生成器实现管道功能
# 输出函数只管输出
def tail(f):
f.seek(0, os.SEEK_END)
while True:
line = f.readline()
if not line:
time.sleep(1)
continue
yield line
# 输入函数只管处理输入内容
def grep(linegtr, searchtxt):
for line in linegtr:
if searchtxt in line:
yield line
# 下面是用户使用代码
f = open('log.txt')
logtxt = tail(f) # logtxt是一个生成器对象，每次
result = grep(logtxt, 'Jerry') # result也是一个生成器对象
for line in result:
print(line)
# 手动关闭生成器
c.close()
# yield语句上捕获GeteratorExit异常
def countdown(n):
try:
while n > 0:
yield n
n -= 1
except GeneratorExit:
print('The current n is %d' %n)
===声明式编程===
lines = open('stationery.txt')
fields = (line.split(',') for lien in lines)
print(sum( float(f[1]) * float(f[2]) for f in fields ))
sum(price*qty for price,qty in
cursor.execute('select price, qty from stationery')
if price*qty >= 100)
===协程===
import asyncio
async def compute(x, y):
print("Compute %s + %s ..." % (x, y))
await asyncio.sleep(1.0)
return x + y
async def print_sum(x, y):
result = await compute(x, y)
print("%s + %s = %s" % (x, y, result))
loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
print("end")
loop.close()
输出:
Compute 1 + 2 ...
1 + 2 = 3
end
import asyncio
future = asyncio.Future()
async def coro1():
print("wait 1 second")
await asyncio.sleep(1)
print("set_result")
future.set_result('data')
async def coro2():
result = await future
print(result)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait([
coro1()
coro2()
]))
loop.close()
输出:
wait 1 second
set_result
data
===多进程===
import multiprocessing
print("Number of cpu : ", multiprocessing.cpu_count())
from multiprocessing import Process
def print_func(continent='Asia'):
print('The name of continent is : ', continent)
if __name__ == "__main__":
names = ['America', 'Europe', 'Africa']
procs = []
proc = Process(target=print_func)
procs.append(proc)
proc.start()
for name in names:
proc = Process(target=print_func, args=(name,))
procs.append(proc)
proc.start()
for proc in procs:
proc.join()
进程池:
from multiprocessing import Pool
import time
work = (["A", 5], ["B", 2], ["C", 1], ["D", 3])
def work_log(work_data):
print(" Process %s waiting %s seconds" % (work_data[0], work_data[1]))
time.sleep(int(work_data[1]))
print(" Process %s Finished." % work_data[0])
def pool_handler():
p = Pool(2)
p.map(work_log, work)
if __name__ == '__main__':
pool_handler()
元类metaclass
一旦你把一个类型 MyClass 的 metaclass 设置成 MyMeta，MyClass 就不再由原生的 type创建，而是会调用 MyMeta 的__call__运算符重载。
class = type(classname, superclasses, attributedict)
这里等号右边的type(classname, superclasses, attributedict)，就是 type 的__call__运算符重载，它会进一步调用：
type.__new__(typeclass, classname, superclasses, attributedict)
type.__init__(class, classname, superclasses, attributedict)
----------------------------------
class = type(classname, superclasses, attributedict)
# 变为了
class = MyMeta(classname, superclasses, attributedict)
class MyMeta(type):
def __new__(cls, *args, **kwargs):
print('===>MyMeta.__new__')
print(cls.__name__)
return super().__new__(cls, *args, **kwargs)
def __init__(self, classname, superclasses, attributedict):
super().__init__(classname, superclasses, attributedict)
print('===>MyMeta.__init__')
print(self.__name__)
print(attributedict)
print(self.tag)
def __call__(self, *args, **kwargs):
print('===>MyMeta.__call__')
obj = self.__new__(self, *args, **kwargs)
self.__init__(self, *args, **kwargs)
return obj
class Foo(object, metaclass=MyMeta):
tag = '!Foo'
def __new__(cls, *args, **kwargs):
print('===>Foo.__new__')
return super().__new__(cls)
def __init__(self, name):
print('===>Foo.__init__')
self.name = name
print('test start')
foo = Foo('test')
print('test end')
元类实现单例
class SingletonMeta(type):
def __init__(self,*args,**kwargs):
self.__instance = None #这是一个私有属性来保存属性，而不会污染Singleton类（其实还是会污染，只是无法直接通过__instance属性访问）
super().__init__(*args,**kwargs)
def __call__(self, *args, **kwargs):
if self.__instance is None:
self.__instance = super().__call__(*args, **kwargs)
return self.__instance
class Singleton(metaclass=SingletonMeta): #在代码执行到这里的时候，元类中的__new__方法和__init__方法其实已经被执行了，而不是在Singleton实例化的时候执行。且仅会执行一次。
pass
class SubSingleton(metaclass=SingletonMeta):
pass
s1 = Singleton()
s2 = Singleton()
print(s1 is s2)
ss1 = SubSingleton()
ss2 = SubSingleton()
print(ss1 is ss2)
Python之数据序列化（json、pickle、shelve）
>>> xInstance = X(1, 2, 3) >>> xInstance(1,2) _call_ with (1, 2)
SingleTon
三种方式实现
1.Module-level singleTon
2.Classic singleTon
3.Borg SingleTon
1.Module-level:
import 天然支持singleton(All moduels are singleton)
# singleton.py
shared_var = 'variable''
# module1.py
import singletom
print(singletom.shared_var)
singletom.shared_var += '(modified by module1)'
# module2.py
import singletom
print(singletom.shared_var)
Run:
import module1.py
import module2.py
输出：
>>> variable
>>> variable(modified by module1)
2.Classic
class SingletonClass(object):
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super(SingletonClass, cls).__new__(cls)
return cls.instance
class SingletonChild(SingletonClass):
pass
singleton = SingletonClass()
child = SingletonChild()
print(singleton is child)
singleton.variable = 'singleton variable'
print(child.variable)
输出(相同的实例，相同的变量)：
True
singleton variable
3.Borg
Borg singleton是python的一种设计模式，允许不同的实例共享相同的变量
class BorgSingleton(object):
_shared_borg_state = {}
def __new__(cls, *args, **kwargs):
obj = super(BorgSingleton, cls).__new__(cls, *args, **kwargs)
obj.__dict__ = cls._shared_borg_state
return obj
borg = BorgSingleton()
borg.shared_variable = "Shared Variable"
class ChildBorg(BorgSingleton):
pass
childBorg = ChildBorg()
print(childBorg is borg)
print(childBorg.shared_variable)
输出：
False
Shareed Variable
_new,init和call,del_
__new__ 是一个静态方法，cls由解释器传值，返回参数
__init__有一个参数self，就是这个__new__返回的实例，__init__在__new__的基础上可以完成一些其它初始化的动作，__init__不需要返回值
若__new__没有正确返回当前类cls的实例，那__init__是不会被调用的，即使是父类的实例也不行
1. __init__(): 初始化某个类的一个实例。
2. __call__(): 使实例能够像函数一样被调用(调用__new__和__init__)
3. __main__(): 命令行执行调用, 先调用__init__
4. __str__(): 显示字串: str(obj), print(obj), format(obj)
5. __repr__(): 显示对象，便于调试: repr(obj)
6. __del__(): 删除实例
7.__getitem__() __setitem__: 调用p['name']
class X(object):
def __new__(cls,*args, **kwargs):
print "new %s"%cls
return object.__new__(cls, *args, **kwargs)
def __init__(self, a, b, range):
self.a = a
self.b = b
self.range = range
def __main__(self):
print('command called!')
def __call__(self, a, b):
self.a = a
self.b = b
print('__call__ with （{}, {}）'.format(self.a, self.b))
def __str__(self):
return print('x object')
def __repr__(self):
return self.__repr__;
def __del__(self, a, b, range):
del self.a
del self.b
del self.range
def __getitem__(self, key):
return self.__dict__.get(key)
def __setitem__(self, key, value):
return self.__dict__.set(key, value)
类变量、实例变量、类方法、实例方法
class Foo:
class_var = '' #类变量，相当于静态变量
def __init__(self):
self._age = 0
self.instance_var = '' #实例变量
def __main__(self): #命令行调用python foo.py
print('usage command line!')
@classmethod
def class_method(cls,parameter):#相当于静态类，cls代表类
print(cls.instance_var)
print(class_var)
@staticmethod
def statis_method(paramter):
print(self.instance_var) #不能访问
print(class_var) #不能访问
def method(self,parameter): #实例类，self代表当前类实例
print(Object.class_var)
print(self,instance_var)
@property
def age(self):
return self._age;
@age.setter
def age(self, value):
if isinstance(value,str):
self._age = Decimal(value)
if isinstance(value, Decimal):
self._age = value
@age.deleter
def age(self):
raise AttributeError("cannot delete attribute")
foo = Foo()
print(foo.age)
foo.age = 12
del foo.age
class A:
v = 22;
a = A()
a.v = 13
print(A.v) #22
print(a.v) #13
del a.v
print(a.v) #22