设计模式

设计模式

设计模式是软件开发人员在软件开发过程中面临的一般问题的解决方案。这些解决方案是众多软件开发人员经过相当长的一段时间的试验和错误总结出来的。

单例模式

最简单的设计模式。控制实例数目，节省系统资源的时候就可以使用单例模式,不过要注意线程安全问题

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
from threading import Lock, Thread
import time


class Singleton:
    __instance = None
    __LOCK = Lock()

    def __new__(cls, *args, **kwargs):
        """懒加载 & 线程安全"""
        if cls.__instance:
            # 先尝试判断是否已经实例化过了，再加锁，提升效率
            time.sleep(1)  # 模拟让出 CPU 的情况
            return cls.__instance

        with cls.__LOCK:
            if cls.__instance:
                # double check 防止实例化两次
                time.sleep(1)  # 模拟让出 CPU 的情况
                return cls.__instance

            # 实例化
            cls.__instance = object.__new__(cls)
        return cls.__instance


def create_singleton():
    print(Singleton())


if __name__ == '__main__':
    # 并发条件下
    t1 = Thread(target=create_singleton)
    t2 = Thread(target=create_singleton)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
PYTHON

再根据 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
from threading import Lock, Thread
import time


def singleton(cls):
    # 创建锁
    cls.__LOCK = Lock()
    cls.__instance = None

    def new(cls, *args, **kwargs):
        if cls.__instance:
            # 先判断是否已经实例化，提升效率
            time.sleep(1)  # 模拟让出 CPU 的情况
            return cls.__instance

        with cls.__LOCK:
            if cls.__instance:
                # double check 防止实例化两次
                time.sleep(1)  # 模拟让出 CPU 的情况
                return cls.__instance

            # 实例化
            cls.__instance = object.__new__(cls)
        return cls.__instance
    # 设置好方法
    cls.__new__ = new
    return cls


@singleton
class Singleton:
    pass
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class Subject:
    def add_observer(self, observer):
        """添加观察者"""
        raise NotImplementedError()

    def remove_observer(self, observer):
        """移除指定的观察者"""
        raise NotImplementedError()

    def notify_all(self):
        """通知所有观察者"""
        raise NotImplementedError()


class Observer:
    def on_update(self):
        raise NotImplementedError()


class NewsPaper(Subject):
    def __init__(self):
        self.obs = set()
        self.__news = ""

    def add_observer(self, observer):
        self.obs.add(observer)

    def remove_observer(self, observer):
        self.obs.remove(observer)

    def notify_all(self):
        for ob in self.obs:
            ob.on_update()

    @property
    def news(self):
        return self.__news

    @news.setter
    def news(self, msg):
        self.__news = msg
        # 通知注册了的观察者
        self.notify_all()


class Citizen(Observer):
    def __init__(self, sub, name):
        self.sub = sub
        self.sub.add_observer(self)
        self.name = name

    def on_update(self):
        print(f'{self.name}: 收到信息变化{self.sub.news}')


if __name__ == '__main__':
    sub = NewsPaper()

    obs1 = Citizen(sub, 'a')
    obs2 = Citizen(sub, 'b')

    sub.news = '下雨了'
    sub.news = '吃饭了'

# 输出
# b: 收到信息变化下雨了
# a: 收到信息变化下雨了
# b: 收到信息变化吃饭了
# a: 收到信息变化吃饭了
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
45
46
47
48
49
50
51
52
53
54
55
class PowerSource:
    """电源提供"""

    def get_power(self):
        raise NotImplementedError()


class Battery110V(PowerSource):
    """110V电源"""

    def get_power(self):
        return 110


class Electric:
    _rated_voltage = 0

    def __init__(self, source: PowerSource):
        self.source = source
        # 开始供电
        self.__supply_power()

    def __supply_power(self):
        rv = self.__class__._rated_voltage
        if self.source.get_power() != rv:
            raise ValueError(f'需要电压为 {rv}V 的电源，但提供的是 {self.source.get_power()}V')
        print('电源供给正常!')


class ToyCar(Electric):
    # 玩具车所需电压为 220
    _rated_voltage = 220


class DoubleAdapter(PowerSource):
    """两倍的变压器"""

    def __init__(self, ps: PowerSource):
        self.ps = ps

    def get_power(self):
        return self.ps.get_power() * 2


if __name__ == '__main__':
    # 暂时只有 100v 的电压
    power110 = Battery110V()
    # 使用一个电源适配器
    power220adapter = DoubleAdapter(power110)

    # 出问题
    # car = ToyCar(power110)

    # 完美解决
    car = ToyCar(power220adapter)
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
class Database:
    def do_query(self):
        raise NotImplementedError()


class MysqlDatabase(Database):
    def do_query(self):
        print('mysql: do_query()')


class OracleDatabase(Database):
    def do_query(self):
        print('oracle: do_query()')


class MongoDatabase(Database):
    def do_query(self):
        print('mongo: do_query()')


class DatabaseFactory:
    __ALL_DB = {
        'mysql': MysqlDatabase,
        'oracle': OracleDatabase,
        'mongo': MongoDatabase
    }

    def get_db(self, db_type):
        DB = self.__class__.__ALL_DB.get(db_type)
        return DB() if DB else None


if __name__ == '__main__':
    factory = DatabaseFactory()

    msql = factory.get_db('mysql')
    oracle = factory.get_db('oracle')
    mongo = factory.get_db('mongo')

    msql.do_query()
    oracle.do_query()
    mongo.do_query()
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
class Database:
    def do_query(self):
        raise NotImplementedError()


class MysqlDatabase(Database):
    def do_query(self):
        print('mysql: do_query()')


class OracleDatabase(Database):
    def do_query(self):
        print('oracle: do_query()')


class DatabaseFactory:
    def create_db(self) -> Database:
        raise NotImplementedError()


class MysqlFactory(DatabaseFactory):
    def create_db(self) -> Database:
        return MysqlDatabase()


class OracleFactory(DatabaseFactory):
    def create_db(self) -> Database:
        return OracleDatabase()


if __name__ == '__main__':
    msql_factory = MysqlFactory()
    oracle_factory = OracleFactory()

    mysql = msql_factory.create_db()
    oracle = oracle_factory.create_db()
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class Vehicle:
    def move(self):
        raise NotImplementedError()


class Food:
    def eat(self):
        raise NotImplementedError()


# ================ Land World ============
class Car(Vehicle):
    def move(self):
        print("car's running...")


class Bike(Vehicle):
    def move(self):
        print("bike's running...")


class Rice(Food):
    def eat(self):
        print("the rice was eaten...")


class Hamburger(Food):
    def eat(self):
        print("the hamburger was eaten...")
# ================ End Land World ============


# ================ Water World ============
class Submarine(Vehicle):
    def move(self):
        print("submarine's swimming...")


class Boat(Vehicle):
    def move(self):
        print("boat's swimming...")


class Fish(Food):
    def eat(self):
        print("the fish was eaten...")


class Crab(Food):
    def eat(self):
        print("the crab was eaten...")
# ================ End Water World ============


class WorldFactory:
    def get_food(self, food_name) -> Food:
        raise NotImplementedError()

    def get_vehicle(self, vehicle_name) -> Vehicle:
        raise NotImplementedError()


class LandWorldFactory(WorldFactory):
    def get_food(self, food_name):
        if food_name == 'rice':
            return Rice()
        elif food_name == 'hamburger':
            return Hamburger()
        return None

    def get_vehicle(self, vehicle_name):
        if vehicle_name == 'bike':
            return Bike()
        elif vehicle_name == 'car':
            return Car()
        return None


class WaterWorldFactory(WorldFactory):
    def get_food(self, food_name):
        if food_name == 'fish':
            return Fish()
        elif food_name == 'crab':
            return Crab()
        return None

    def get_vehicle(self, vehicle_name):
        if vehicle_name == 'submarine':
            return Submarine()
        elif vehicle_name == 'boat':
            return Boat()
        return None


class FactoryProducer:
    @classmethod
    def get_world_factory(cls, world_type):
        if world_type == 'land':
            return LandWorldFactory()
        if world_type == 'water':
            return WaterWorldFactory()
        return None


if __name__ == '__main__':
    factory = FactoryProducer.get_world_factory('land')
    food1 = factory.get_food('rice')
    food2 = factory.get_food('hamburger')
    food1.eat()
    food2.eat()
    v1 = factory.get_vehicle('car')
    v2 = factory.get_vehicle('bike')
    v1.move()
    v2.move()
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
from collections import Iterable


class Readable:
    def read(self):
        raise NotImplementedError()


class IterReader(Readable):
    def __init__(self, iter: Iterable):
        self.iter = iter.__iter__()
        self.tmp = None

    def read(self):
        return self.iter.__next__()


class BufferedIterReader(Readable):
    def __init__(self, reader: Readable):
        self.reader = reader

    def read(self):
        """直接使用被包装的实现"""
        return self.reader.read()

    def read_line(self):
        res, cur = [], self.read()
        while cur != '\n':
            res.append(cur)
            cur = self.read()
        return ''.join(res)


if __name__ == '__main__':
    r = IterReader('abc\ndefg\ndfdsfasdf\n')
    b = BufferedIterReader(r)
    print(b.read_line())
    print(b.read_line())
    print(b.read_line())

# 输出
# abc
# defg
# dfdsfasdf
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import time


class Filter:
    def before(self, func_name, *args, **kwargs):
        pass

    def after(self, func_name, *args, **kwargs):
        pass


class Logger(Filter):
    def before(self, func_name, *args, **kwargs):
        print(f"[logger] '{func_name}' called with params: {args} {kwargs}")


class Timer(Filter):

    def __init__(self):
        self.start = 0

    def before(self, func_name, *args, **kwargs):
        self.start = time.time()

    def after(self, func_name, *args, **kwargs):
        print(f'[timer] function cost: {time.time() - self.start}')


class FilterChain(Filter):
    def __init__(self, *fs):
        self.__filters = fs

    def filters(self, func):
        def decorator(*args, **kwargs):
            self.before(func.__name__, *args, **kwargs)
            res = func(*args, **kwargs)
            self.after(func.__name__, *args, **kwargs)
            return res
        return decorator

    def after(self, func_name, *args, **kwargs):
        for f in self.__filters:
            f.after(func_name, *args, **kwargs)

    def before(self, func_name, *args, **kwargs):
        for f in self.__filters:
            f.before(func_name, *args, **kwargs)


if __name__ == '__main__':
    filter_chain = FilterChain(Logger(), Timer())

    @filter_chain.filters
    def func(a, b):
        print("i'm func.")
        return a + b

    print(func(1, 2))
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
45
46
class TV:
    def open(self):
        print('开灯')

    def close(self):
        print('关电视')


class Light:
    def open(self):
        print('开电视')

    def close(self):
        print('关灯.')


class Computer:
    def open(self):
        print('开电脑')

    def close(self):
        print('关电脑')


class Manager:
    def __init__(self):
        self.tv = TV()
        self.light = Light()
        self.computer = Computer()

    def open(self):
        self.tv.open()
        self.light.open()
        self.computer.open()

    def close(self):
        self.tv.close()
        self.light.close()
        self.computer.close()


if __name__ == '__main__':
    manger = Manager()

    manger.close()
    manger.open()
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
class Strategy:
    def do(self, a, b):
        raise NotImplementedError()


class AddStrategy(Strategy):
    def do(self, a, b):
        return a+b


class SubStrategy(Strategy):
    def do(self, a, b):
        return a - b


class Operator:
    def __init__(self, strategy=None):
        self.__stg = strategy
        if not strategy:
            self.__stg = AddStrategy()

    @property
    def strategy(self):
        return self.__stg

    @strategy.setter
    def strategy(self, stg: Strategy):
        self.__stg = stg

    def do_operator(self, a, b):
        return self.__stg.do(a, b)


if __name__ == '__main__':
    op = Operator()

    op.strategy = AddStrategy()
    print(op.do_operator(1, 2))

    op.strategy = SubStrategy()
    print(op.do_operator(1, 2))
PYTHON

代理模式

比如对象创建开销很大，或者某些操作需要安全控制，或者需要进程外的访问，直接访问会给使用者或者系统结构带来很多麻烦，我们可以在访问此对象时加上一个对此对象的访问层

以下是 Cache代理

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
class Message:
    @property
    def msg(self):
        raise NotImplementedError()


class RemoteMessage(Message):
    def __init__(self, location):
        self.__uri = location
        self.__msg = self.__do_request()

    def __do_request(self):
        print(f'request from {self.__uri}')
        return f"msg from {self.__uri}"

    @property
    def msg(self):
        return self.__msg


class CachedMessage(Message):
    def __init__(self, location):
        self.__uri = location
        self.__msg = None

    @property
    def msg(self):
        if self.__msg is None:
            self.__msg = RemoteMessage(self.__uri)
        return self.__msg.msg


if __name__ == '__main__':
    msg = CachedMessage('127.0.0.1/test')
    print(msg.msg)
    print(msg.msg)
    print(msg.msg)
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
from abc import ABCMeta, abstractmethod
class Iterator(metaclass=ABCMeta):
    @abstractmethod
    def has_next(self) -> bool:
        pass
    @abstractmethod
    def next(self):
        pass

class Collection(Iterator):
    def has_next(self) -> bool:
        return self.index < len(self.args)
    def next(self):
        if not self.has_next():
            raise ValueError('No element remained.')
        res = self.args[self.index]
        self.index += 1
        return res
    def __init__(self, *args):
        self.args = args
        self.index = 0
    def iterator(self) -> Iterator:
        return self

if __name__ == '__main__':
    c = Collection(*[1, 2, 3, 4, 5])
    while c.has_next():
        print(c.next())
PYTHON

使用 python 自带的迭代机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Collection:
    def __init__(self, *args):
        self.args = args
        self.index = 0
    def __next__(self):
        if self.index >= len(self.args):
            raise StopIteration()
        res = self.args[self.index]
        self.index += 1
        return res
    def __iter__(self):
        return self

if __name__ == '__main__':
    for n in Collection(*[1, 2, 3, 4, 5]):
        print(n)
PYTHON

状态模式

状态模式和策略模式非常的相似，不同点是，状态的转换是自动，无意识的，而策略模式的策略变换是手动，主观的去改变的

有一个场景，模拟食堂吃饭的不同状态和不同的行为

  • 给饭
    • 没付钱状态: 请先给钱
    • 付了钱状态: 开始做饭,请等待
    • 在做饭状态: 在做饭了，别催
    • 售罄状态: 没饭了再见
  • 收钱
    • 没付钱状态: 付钱成功!
    • 付了钱状态: 请不要重复付钱
    • 在做饭状态: 在做饭了，别给钱了
    • 售罄状态: 没饭了，别给了
  • 退钱
    • 没付钱状态: 没付钱还想退钱？
    • 付了钱状态: 退款成功
    • 在做饭状态: 都在做饭,不能退
    • 售罄状态: 你想多了
  • 发饭
    • 没付钱状态: 请先去买饭
    • 付了钱状态:
    • 在做饭状态: 好了好了给你饭
    • 售罄状态: 没饭给你了

经过简单思考一后，有可能写出以下代码:

1
2


PYTHON
设计模式
#设计模式 #技巧
设计模式
https://blog.mjhxyz.top/posts/41682/
作者
Mao
发布于
2020年10月9日
许可协议