十一、python实现单例模式

---

单例模式以及Python实现

Python单例模式(Singleton)的N种实现

1、单例模式介绍

单例模式就是确保一个类只有一个实例.当你希望整个系统中,某个类只有一个实例时,单例模式就派上了用场.
比如,某个服务器的配置信息存在在一个文件中,客户端通过AppConfig类来读取配置文件的信息.如果程序的运行的过程中,很多地方都会用到配置文件信息,则就需要创建很多的AppConfig实例,这样就导致内存中有很多AppConfig对象的实例,造成资源的浪费.其实这个时候AppConfig我们希望它只有一份,就可以使用单例模式.

2、实现单例模式的几种方法

2.1 使用模块

python的模块就是天然的单例模式,因为模块在第一次导入的时候,会生成.pyc文件,当第二次导入的时候,就会直接加载.pyc文件,而不是再次执行模块代码.如果我们把相关的函数和数据定义在一个模块中,就可以获得一个单例对象了.

• 代码

class Singleton(object):
    def foo(self):
        pass
singleton = Singleton()
1
2
3
4

• 使用

from singleton.mysingleton import singleton
1

2.2 使用装饰器

装饰器里面的外层变量定义一个字典,里面存放这个类的实例.当第一次创建的收,就将这个实例保存到这个字典中.
然后以后每次创建对象的时候,都去这个字典中判断一下,如果已经被实例化,就直接取这个实例对象.如果不存在就保存到字典中.

# encoding:utf-8
__author__ = 'Fioman'
__time__ = '2019/3/6 10:22'
 
 
def singleton(cls):
    # 单下划线的作用是这个变量只能在当前模块里访问,仅仅是一种提示作用
    # 创建一个字典用来保存类的实例对象
    _instance = {}
 
    def _singleton(*args, **kwargs):
        # 先判断这个类有没有对象
        if cls not in _instance:
            _instance[cls] = cls(*args, **kwargs)  # 创建一个对象,并保存到字典当中
        # 将实例对象返回
        return _instance[cls]
 
    return _singleton
 
 
@singleton
class A(object):
    a = 1
 
    def __init__(self, x=0):
        self.x = x
        print('这是A的类的初始化方法')
 
 
a1 = A(2)
a2 = A(3)
print(id(a1), id(a2))
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

2.3 使用类

思路就是,调用类的instance方法,这样有一个弊端就是在使用类创建的时候,并不是单例了.也就是说在创建类的时候一定要用类里面规定的方法创建
会有问题

# encoding:utf-8
__author__ = 'Fioman'
__time__ = '2019/3/6 11:06'
 
 
class Singleton(object):
    def __init__(self,*args,**kwargs):
        pass
 
    @classmethod
    def get_instance(cls, *args, **kwargs):
        # 利用反射,看看这个类有没有_instance属性
        if not hasattr(Singleton, '_instance'):
            Singleton._instance = Singleton(*args, **kwargs)
 
        return Singleton._instance
 
 
s1 = Singleton()  # 使用这种方式创建实例的时候,并不能保证单例
s2 = Singleton.get_instance()  # 只有使用这种方式创建的时候才可以实现单例
s3 = Singleton()
s4 = Singleton.get_instance()
 
print(id(s1), id(s2), id(s3), id(s4))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

2.4 基于__new__方法实现的单例模式(推荐使用,方便)

1、一个对象的实例化过程是先执行类的__new__方法,如果我们没有写,默认会调用object的__new__方法,返回一个实例化对象,然后再调用__init__方法,对这个对象进行初始化,我们可以根据这个实现单例.
2、在一个类的__new__方法中先判断是不是存在实例,如果存在实例,就直接返回,如果不存在实例就创建.

# encoding:utf-8
__author__ = 'Fioman'
__time__ = '2019/3/6 13:36'
import threading
 
 
class Singleton(object):
    _instance_lock = threading.Lock()
 
    def __init__(self, *args, **kwargs):
        pass
 
    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, '_instance'):
            with Singleton._instance_lock:
                if not hasattr(cls, '_instance'):
                    Singleton._instance = super().__new__(cls)
 
            return Singleton._instance
 
 
obj1 = Singleton()
obj2 = Singleton()
print(obj1, obj2)
 
 
def task(arg):
    obj = Singleton()
    print(obj)
 
 
for i in range(10):
    t = threading.Thread(target=task, args=[i, ])
    t.start()
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

2.5 使用函数装饰器实现单例

def singleton(cls):
    _instance = {}

    def inner():
        if cls not in _instance:
            _instance[cls] = cls()
        return _instance[cls]
    return inner
    
@singleton
class Cls(object):
    def __init__(self):
        pass

cls1 = Cls()
cls2 = Cls()
print(id(cls1) == id(cls2))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3、实战

Python 中的单例

3.1 简单实现

class MusicPlayer(object):

    # 记录第一个被创建对象的引用
    instance = None

    # 记录是否执行过初始化动作
    init_flag = False

    def __new__(cls, *args, **kwargs):

        # 判断类属性是否是空对象
        if cls.instance is None:

            # 调用父类的方法，为第一个对象分配空间
            cls.instance = super().__new__(cls)
        # 返回类属性保存的对象引用
        return cls.instance

    def __init__(self):
        # 判断是否执行过初始化动作
        if MusicPlayer.init_flag:
            return
        # 如果没有执行过，在执行初始化动作
        print("初始化播放器")
        # 修改类属性的标记
        MusicPlayer.init_flag = True

# 创建多个对象
play1 = MusicPlayer()
print(play1)

play2 = MusicPlayer()
print(play2)

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

3.2 实际应用配置文件

import configparser
import json
import os
import requests
from base import baselog

region_file = 'apiserver.conf'
CONFIG_PATH = os.path.dirname(os.path.abspath(__file__))
file = os.path.join(CONFIG_PATH, region_file)
domain_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'domain.conf')
log = baselog.PyLogger().logger

class Config(object):

    # 记录第一个被创建对象的引用
    instance = None

    # 记录是否执行过初始化动作
    init_flag = False
    config = configparser.ConfigParser()
    if os.path.isfile(domain_file):
        config.read(domain_file)
        rds_host = config.get('rds', 'rds_host')
        rds_endpoint = config.get('rds', 'rds_endpoint')

        vpc_host = config.get('vpc', 'vpc_host')
        vpc_endpoint = config.get('vpc', 'vpc_endpoint')

        ak = config.get('account', 'ak_normal')
        sk = config.get('account', 'sk_normal')
        log.info('rds_host: {0}'.format(rds_host))
        log.info('rds_endpoint: {0}'.format(rds_endpoint))
        log.info('vpc_host: {0}'.format(vpc_host))
        log.info('vpc_endpoint: {0}'.format(vpc_endpoint))
        log.info('ak: {0}'.format(ak))
        log.info('sk: {0}'.format(sk))
    else:
        config.read(file)
        ak = config.get('account', 'ak_normal')
        res = {}
        # 判断ic接口
        ic_flage = True
        try:
            ednpoint = config.get('ic', 'api_gateway_url')
            result = requests.get(ednpoint)
            res1 = json.loads(result.text)
            log.info('test endpoint ic: {0}'.format(res1))

            account = config.get('ic', 'iam_url')
            result = requests.get(account)
            res2 = json.loads(result.text)
            log.info('test account ic: {0}'.format(res2))
        except Exception as e:
            ic_flage = False
            log.warning('test get ic failed: {0}'.format(e))

        if ak == 'None' and ic_flage:
            try:
                ednpoint = config.get('ic', 'api_gateway_url')
                result = requests.get(ednpoint)
                res = json.loads(result.text)['data']['api_gateway'][0]
            except Exception as e:
                log.warning('get ic failed: {0}'.format(e))
                log.info('use default.conf: {0}'.format("default"))
                default_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "default.conf")
                with open(default_file, "r") as fp:
                    res = json.load(fp)["data"]['api_gateway'][0]

            # get domain
            try:
                rds_domain = res['api-logic-rds_dns_name']
                bcc_domain = res['bcc_domain']
                region = res['region']
            except Exception as e:
                log.warning('get ednpoint failed:{0}'.format(e))
            # get paas as sk
            try:
                account = config.get('ic', 'iam_url')
                result = requests.get(account)
                res = json.loads(result.text)['data']['iam_nginx'][0]
            except Exception as e:
                log.warning('get ic failed: {0}'.format(e))
                log.info('use default.conf: {0}'.format("default"))
                default_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "default.conf")
                with open(default_file, "r") as fp:
                    res = json.load(fp)["data"]['iam_nginx'][0]
            try:
                ak = res['paas_user_1_ak']
                sk = res['paas_user_1_sk']
            except Exception as e:
                log.warning('get account failed:{0}'.format(e))

            # 配置文件重置
            try:
                config.set("account", "ak_normal", ak)
                config.set("account", "sk_normal", sk)
                config.set("rds", "rds_host", rds_domain + ':8680')
                config.set("rds", "rds_endpoint", rds_domain + ':8680')
                config.set("vpc", "vpc_host", bcc_domain + ':8680')
                config.set("vpc", "vpc_endpoint", bcc_domain + ':8680')
                config.set("region", "region", region)
                config.write(open(domain_file, 'w'))
                config.read(domain_file)
                rds_host = config.get('rds', 'rds_host')
                rds_endpoint = config.get('rds', 'rds_endpoint')

                vpc_host = config.get('vpc', 'vpc_host')
                vpc_endpoint = config.get('vpc', 'vpc_endpoint')

                ak = config.get('account', 'ak_normal')
                sk = config.get('account', 'sk_normal')
                log.info('rds_host: {0}'.format(rds_host))
                log.info('rds_endpoint: {0}'.format(rds_endpoint))
                log.info('vpc_host: {0}'.format(vpc_host))
                log.info('vpc_endpoint: {0}'.format(vpc_endpoint))
                log.info('ak: {0}'.format(ak))
                log.info('sk: {0}'.format(sk))
            except Exception as e:
                log.warning('can not get account、 host:{0}'.format(e))
        else:
            # 可手动适配环境
            config.read(file)
            rds_host = config.get('rds', 'rds_host')
            rds_endpoint = config.get('rds', 'rds_endpoint')

            vpc_host = config.get('vpc', 'vpc_host')
            vpc_endpoint = config.get('vpc', 'vpc_endpoint')

            ak = config.get('account', 'ak_normal')
            sk = config.get('account', 'sk_normal')
            log.info('rds_host: {0}'.format(rds_host))
            log.info('rds_endpoint: {0}'.format(rds_endpoint))
            log.info('vpc_host: {0}'.format(vpc_host))
            log.info('vpc_endpoint: {0}'.format(vpc_endpoint))
            log.info('ak: {0}'.format(ak))
            log.info('sk: {0}'.format(sk))

    def __new__(cls, *args, **kwargs):

        # 判断类属性是否是空对象
        if cls.instance is None:
            # 调用父类的方法，为第一个对象分配空间
            cls.instance = super().__new__(cls)
        # 返回类属性保存的对象引用
        return cls.instance

    def __init__(self):
        # 判断是否执行过初始化动作
        if Config.init_flag:
            return
        # 如果没有执行过，在执行初始化动作
        # 修改类属性的标记
        Config.init_flag = True


if __name__ == '__main__':
    print(Config().rds_host)
    print(Config().rds_host)
    print(Config().ak)
    print(Config().sk)
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160