Python杂项

发表于 | 更新于

常识

Convention

__foo__: 这只是一种约定，是Python系统使用不会与用户名冲突的名称的一种方式。

__foo: 这具有实际意义：解释器将此名称替换_classname__foo为确保名称不会与另一个类中的类似名称重叠的方式。

_foo: 这只是一种约定，是程序员指示变量是私有的一种方式（无论在Python中是什么意思）。

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
32
33
34
35
36
37
38
39
40
41
42
43
44
class DoubleLine(object):
    """
    双下划线的类变量，实例变量会自动转换成_classname__variable的形式。
    """
    __class_variable = 100

    def __init__(self):
        self.__instance_variable = 101
        self.__setattr__('_DoubleLine__local', 102)
        object.__setattr__(self, '_DoubleLine__local2', 103)

    def to_dict(self):
        return self.__dict__

    def list_all_member(self):
        for name, value in vars(self).items():
            print('%s=%s' % (name, value))

    def dir_all_attr_and_func(self):
        for value in dir(self):
            print(value)

    def test(self):
        print(self.__local)
        print(self.__local2)
        print(self.__class_variable)
        print(self.__instance_variable)
        try:
            print(self.__localdynamic_add)
            print(self._DoubleLine__local2)
        except Exception as exception:
            print(exception)


dl = DoubleLine()
dl._DoubleLine__localdynamic_add = 99
print(dl.to_dict())

dl.list_all_member()
dl.dir_all_attr_and_func()
dl.test()
print(dl._DoubleLine__class_variable)
print(dl._DoubleLine__instance_variable)
print(dl.__instance_variable) # 异常提示属性不存在。

线程

Py3中原始的线程引用_thread模块。

local 线程局部变量。

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
from threading import local
import time
import _thread as thread  # 等于在PY2中的原始线程。

def print_time(threadName, delay):
    # 为线程定义一个函数，线程局部变量。
    thread_local_data = local()
    thread_local_data.count = 0
    count = 0
    while count < 5:
        time.sleep(delay)
        count += 1
        thread_local_data.count += 1
        print("%s: %s count: %s" % (threadName, time.ctime(time.time()), thread_local_data.count))
    print("%s: count: %s exit" % (threadName, thread_local_data.count))

def empty_thread():
    print("empty_thread")

# 创建三个线程
try:
    thread.start_new_thread(empty_thread, ())
    thread.start_new_thread(print_time, ("Thread-1", 2))
    thread.start_new_thread(print_time, ("Thread-2", 3))
except:
    print("Error: unable to start thread")

while 1:
    time.sleep(1)

python 默认参数创建线程后，不管主线程是否执行完毕，都会等待子线程执行完毕才一起退出，有无join结果一样

如果创建线程，并且设置了daemon为true，及threading.setDaemon(True),则主线程执行完毕后自动退出，不会等待子线程的执行结果。而且随着主线程的退出，子线程也消亡

join方法的作用是阻塞，等待子线程结束，join方法有一个参数是timeout，即如果主线程等待timeout，子线程还没有结束，则主线程强制结束子线程。

如果线程daemon属性为False，则join里的timeout参数无效，主线程会一直等待子线程结束。

如果线程daemon属性为True，则join里的timeout参数是有效的，主线程会等待timeout时间后，结束子线程。

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
from threading import local, Thread
import time
import random

exitFlag = False

class TaskThread(Thread):

    def __init__(self, thread_name, counter, delay):
        Thread.__init__(self)
        self.thread_name = thread_name
        self.counter = counter
        self.delay = delay

    def run(self):
        print("Starting " + self.thread_name)
        self.internal_run()
        print("Exiting " + self.thread_name)

    def internal_run(self):
        global exitFlag
        while self.counter:
            if exitFlag:
                break
            time.sleep(self.delay)
            print("%s: %s" % (self.thread_name, time.ctime(time.time())))
            self.counter -= 1

threads = []
thread1 = TaskThread("Thread-1", 5, 1)
thread2 = TaskThread("Thread-2", 5, 2)

thread1.start()
thread2.start()
threads.append(thread1)
threads.append(thread2)

time.sleep(3)
exitFlag = True

for item in threads:
    item.join()
print("Exit")

def doThreadTest(delay):
    print('start thread time:', time.strftime('%H:%M:%S'))
    print(delay)
    time.sleep(delay)
    print('stop thread time:', time.strftime('%H:%M:%S'))

threads = []
for i in range(3):
    t = Thread(target=doThreadTest, args=(random.randint(1, 5), ))
    # t.setDaemon(True) # 设置后主线程退出，子线程跟随退出。
    threads.append(t)

for t in threads:
    t.start()

for t in threads:
    t.join(1)
print('stop main thread')

互斥锁

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
32
33
34
35
import threading
import os
import time

class Ticket():
    def __init__(self, count):
        self.ticket = {'ticket': count}

    def check_ticket(self):
        print('剩余{}票'.format(self.ticket['ticket']))

    def deal_ticket(self):
        if self.ticket['ticket'] > 0:
            self.ticket['ticket'] -= 1
            print('%s购票成功' % threading.current_thread().ident)
        else:
            print('%s购票失败' % threading.current_thread().ident)

    def buy_ticket(self, locks):
        locks.acquire()
        self.check_ticket()
        self.deal_ticket()
        locks.release()


locks = threading.Lock()
ticket = Ticket(10)
threads = []
for i in range(11):
    t = threading.Thread(target=ticket.buy_ticket, args=(locks,))
    t.start()
    threads.append(t)

for item in threads:
    item.join()

信号量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import threading
import time
count = threading.active_count()

def run(n):
    semaphore.acquire()
    time.sleep(1)
    print('run the thread:%s\n' % n)
    semaphore.release()

semaphore = threading.BoundedSemaphore(5)
for i in range(20):
    threading.Thread(target=run, args=(i,)).start()

while threading.active_count() != count:
    time.sleep(1)
else:
    print('over')

事件

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
32
33
34
import time
import threading

event = threading.Event()
event.set()

def Lighter():
    count = 0
    while True:
        if count > 10 and count < 20:
            event.clear()
            print('Red Light')
        elif count > 20:
            count = 0
            event.set()
        else:
            print('Green Light')
        time.sleep(1)
        count += 1

def Car():
    while True:
        if event.is_set():
            print('Running..')
            time.sleep(1)
        else:
            print('Red Light Waitting')
            event.wait()

light = threading.Thread(target=Lighter)
car = threading.Thread(target=Car)
light.start()
car.start()
light.join()

类常识

  1. 常识，静态语言不能支持2个相同名字的变量。
1
2
3
4
5
6
7
8
9
class Attr(object):
    class_variable = "类变量"  # 等价于Class中申明Static变量
    def __init__(self):
        self.attribute_variable = "属性变量"  # 等价于成员变量
        self.class_variable = "属性变量-冲突测试"  # 当没有定义__slots__的时候不会冲突。

a = Attr()
print(a.class_variable)  # 属性变量-冲突测试
print(Attr.class_variable)  # 类变量，访问静态成员变量，这个也是静态语言不能实现的。

冲突，具体跟Python针对slots的实现处理方式有关。

1
2
3
4
5
6
7
class Attr(object):
    #__slots__ = ('age')
    __slots__ = () # 只要有定义就会冲突。

Attr.age = 99 # 绑定类变量。
a = Attr()
a.age = 100  # 'Attr' object attribute 'age' is read-only 当定义了__slots__后，不能和类变量冲突。
  1. __slots__ 定义和类变量冲突问题。
1
2
3
class Attr(object):
    class_variable = "类变量"  # 等价于Class中申明Static变量
    __slots__ = ('class_variable') # 'class_variable' in __slots__ conflicts with class variable 冲突。
1
2
3
4
5
6
class Attr
{
public:
	static std::string g_class_variable;
	std::string g_class_variable; //编译提示重定义。
};
  1. __slots__完成对实例变量绑定的限制。【限制是类的实例-这个说法应该是不精确的】
1
2
3
4
5
6
7
8
class Attr(object):
    class_variable = "类变量"
    __slots__ = ('attr_info')
    def __init__(self):
        # 限制属性变量的绑定, 'Attr' object has no attribute 'attribute_variable'
        self.attribute_variable = "属性变量" 

a = Attr()

类定义外部绑定属性也会同样的问题。

1
2
3
4
5
class Attr(object):
    __slots__ = ()

a = Attr()
a.other = "'Attr' object has no attribute 'other'"
  1. __slots__ 完成对实例成员函数绑定的限制。
1
2
3
4
5
6
7
8
def set_age(self, age):
    self.age = age

class Attr(object):
    __slots__ = ()

a = Attr()
a.set_age = MethodType(set_age, Attr) #'Attr' object has no attribute 'set_age'
  1. __slots__不限制对class全局变量（类变量）绑定的添加。
1
2
3
class Attr(object):
    __slots__ = ()
Attr.set_age = "'Attr' object attribute 'set_age' is read-only"
  1. MethodType绑定到类方法后的行为。
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def set_name(self, name):
    self.name = name

class Attr(object):
    __slots__ = ('set_name')

a = Attr()
# 绑定到类中，执行set_name方法添加的self.name是添加到类变量中。
# Attr.set_name = MethodType(set_name, Attr)
# 绑定到实例方法中去，这个属性绑定会被__slots__的设置限制，
# 执行set_name方法添加的self.name也是同样会添加到类变量中。
# a.set_name = MethodType(set_name, Attr)
# a.set_name('Ewkoll') # 这里会提示异常，因为__slots__中没有name。
# print(a.name)
# print(Attr.name)

# 绑定实例函数到实例上，受到__slots__控制。
a.set_name = MethodType(set_name, a) 
a.set_name('Ewkoll') # 这里会提示异常，因为__slots__中没有name。当__slots__中有name的时候，添加的是成员变量。
print(a.name)
print(Attr.name) #<member 'name' of 'Attr' objects>

# 绑定到类的方法中。不受到__slots__控制。但是name属性的设置只有a实例拥有
Attr.set_name1 = MethodType(set_name, a)

a = Attr()
Attr.set_name = MethodType(set_name, a)
b = Attr()
print(Attr.name)
# print(b.name) # 异常没有属性变量 name
# b.name = "异常
b.set_name('Ewkoll')  # 这里设置的name是类变量。
# print(b.name) # 异常没有属性变量 name

a = Attr()
Attr.set_name = MethodType(set_name, a)
a.set_name('Ewkoll')
print(a.name)
Attr.name = "32"
# a.name = "99" # "异常，类变量name是只读的"
print(Attr.name)
print(a.name)

b = Attr()
print(Attr.name)
print(b.name)
# b.name = "异常，类变量name是只读的"
b.set_name('Ewkoll')  # 异常，类变量name是只读的， a实例可以调用，但是b实例不能调用。

Flask之Local、LocalStack、LocalProxy

Local解决线程TLS（TlsAlloc）存储无法解决协助之间数据隔离问题而创建的。