一些python不基础的基础知识动态的定义类metaclass，元类的使用装饰器的理解关于*和**的分解作用关于*args, **kwargs关于MRO关于GILGIL与multithreading与multiprocessing关于TYPE HINT

134 阅读 0 评论 89 点赞

我是靠谱客的博主炙热白猫，这篇文章主要介绍一些python不基础的基础知识动态的定义类metaclass，元类的使用装饰器的理解关于*和**的分解作用关于*args, **kwargs关于MRO关于GILGIL与multithreading与multiprocessing关于TYPE HINT，现在分享给大家，希望可以做个参考。

动态的定义类

万物皆对象，类本身也是对象，一个类是一个type对象

复制代码

1
2
3
4
5
class A:
pass
print(type(A))
# 输出是 <class `type`>

这意味着我们可以使用type动态的定义一个类

复制代码

1
2
3
4
5
6
def f(self):
self.name = "bob"
dic = { "id":3 , "f":f }
A = type('A',(),dic)
a = A()

以上代码等价于

复制代码

1
2
3
4
5
6
class A():
id = 3
def f(self):
self.name = "bob"
a = A()

self没有任何特殊含义，就是传入f的第一个变量，作为对象方法时，第一个变量自动填充为f前面的对象
type()函数一个参数的时候就是常见的返回对象的类，传入三个参数时用于构建一个新的type对象，也就是新类。第一个参数是新类的名字（这个名字可以与变量名不相等，例如可以改成type('B,(),dic)），第二个参数是新类所继承的父类，第三个参数是新类的__dict__就简单理解为这个类的空间，包含该类的所有成员变量、成员函数

metaclass，元类的使用

元类需要结合上面type的内容来理解，TOBEDONE

装饰器的理解

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def repeat(n):
def function1(func):
def function2(x):
for i in range(n):
func(x)
return function2
return function1
@repeat(3)
def sayhi(name):
print('hi'+name,end=' ')
# above equals
`sayhi=repeat(3)(sayhi)`
sayhi('bob')
# 输出结果是: hibob hibob hibob

print(end = ' ')中的可选参数end决定在每次打印之后最后输出的内容，默认是n换行，这里指定为空格
这里使用装饰器修饰sayhi函数，使得其调用时执行3遍自身
原理 TOBEDONE
装饰器可以修饰的不只是函数，例如可以修饰类，TOBEDONE

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
def insert_method(cls):
def method(self):
print('method insert')
cls.f = method
return cls
@insert_method
class A:
pass
# above equals `A = insert_method(A)`
a=A()
a.f()
# 输出内容 method insert

装饰器自身也可以不是函数，例如装饰器类 TOBEDONE

关于*和**的分解作用

在函数参数中，在[*]列表生成中
TOBEDONE

关于*args, **kwargs

is what? TOBEDONE

关于MRO

TOBEDONE

关于GIL

垃圾回收与GIL
TOBEDONE

GIL与multithreading与multiprocessing

TOBEDONE

复制代码

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
from multiprocessing import Pool, cpu_count
import time
import os
import traceback
def process_task(number):
try:
print("进程id为: %d, 处理的任务为：%d, 进程处理【开始】" % (os.getpid(), number))
time.sleep(3)
print("进程id为: %d, 处理的任务为：%d, 进程处理【结束】" % (os.getpid(), number))
except Exception as e:
print("Exception: " + str(e))
traceback.print_exc()
raise Exception(str(e))
if __name__ == '__main__':
print("cpu数量为：%d" % cpu_count())
print("主进程id为: %d" % os.getpid())
print("进程开始处理了")
# 进程池有3个进程, 进程数量可以大于cpu_count()的数量, 且os.getpid()获取的数值都不一样
process_pool = Pool(15)
for number in range(30):
process_pool.apply_async(process_task, args=(number,))
print("等待所有进程执行完成")
process_pool.close()
process_pool.join()

复制代码

1
2
3
4
5
6
7
8
9
10
11
import os
import time
def fun(index):
print(index,os.getpid())
time.sleep(3)
from concurrent.futures import ThreadPoolExecutor
thread_pool = ThreadPoolExecutor(max_workers=2)
for i in range(10):
thread_pool.submit(fun, i)
thread_pool.shutdown(wait= True)