Python 单例实际应用

引用别人的一段话。个人感觉很清楚.

概述

单例模式保证了在程序的不同位置都可以且仅可以取到同一个对象实例；如果实例不存在，会创建一个实例；如果已存在就会返回这个实例。因为单例是一个类，所以你也可以为其提供相应的操作方法，以便于对这个实例进行管理

场景

比如你开发一款游戏软件，游戏中需要有“场景管理器”这样一种东西，用来管理游戏场景的切换、资源载入、网络连接等等任务。这个管理器需要有多种方法和属性，在代码中很多地方会被调用，且被调用的必须是同一个管理器，否则既容易产生冲突，也会浪费资源。这种情况下，单例模式就是一个很好的实现方法

三种方式
函数装饰器

类装饰器

new

函数装饰器
def singleton(cls):
_instance = {}

def inter():
if cls not in _instance:
_instance[cls] = cls()
return _instance[cls]

return inter

@singleton
class Test:
def __init__(self):
pass
# 验证
if __name__ == ‘__main__’:
t1 = Test()
t2 = Test()
print(id(t1) == id(t2))
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
类装饰器
class Singleton:
instance = {}

def __init__(self, cls):
self.cls = cls
self.instance = {}

def __call__(self):
if self.cls not in self.instance:
self.instance[self.cls] = self.cls()
return self.instance[self.cls]
# 验证同上

new
class Test:
instance = None

def __new__(cls, *args, **kwargs):
if cls.instance is None:
cls.instance = object.__new__(cls)
return cls.instance
# 验证同上

这里需要注意。使用__new__ 并不是一个真正的称谓一个真正的单例。从类的角度思考：实例化类执行顺序 new –> init.所以每次实例化将会重新实例化init。不是实际意义上的单例模式

实际应用
使用
mysql 连接 minio client 都可以封装为一个单例，对外提供一个连接对象或者直接使用连接对象都很方便
连接异常

连接超时存在 mysql 常见连接8小时。可以尝试捕捉此类异常重新获取或一个连接对象。加入尝试次数等等
连接错误，mysql中 execute捕捉此类异常。minio 是一个http服务可以在每个调用连接对象中捕捉异常
minio 代码实例

登录后复制
from minio import Minio
from minio.error import InvalidResponseError

import uuid
import os
import logging

def singleton(cls):
“””单例”””
_instance = {}

def wrapper(*args, **kwargs):
if cls not in _instance:
_instance[cls] = cls(*args, **kwargs)
return _instance[cls]

return wrapper

def exc_handler(f):
“””异常处理”””

def wrapper(*args, **kwargs):
try:
return f(*args, **kwargs)
except InvalidResponseError as e:
logging.error(e)
# 修改异常返回结果
return False

return wrapper

def read_file(file_path, chunk_size=512):
“””读取”””

with open(file_path, ‘rb’, encoding=’utf-8′) as f:

while True:
chunk = f.read(chunk_size)
if not chunk:
return
yield chunk

def write_file(save_path, data, chunk_size):
“””
文件保存到本地
:param save_path: 文件路径和名称
:param data: 文件流。urllib3.response.HTTPResponse
:param chunk_size:
:return:
“””
with open(save_path, ‘wb’) as file_data:
for d in data.stream(chunk_size):
file_data.write(d)

@singleton
class MINioClient:
“””单例 minio client”””
client = None

def __new__(cls, *args, **kwargs):
if cls.client is None:
cls.client = Minio(
endpoint=kwargs[‘endpoint’],
access_key=kwargs[‘access_key’],
secret_key=kwargs[‘secret_key’],
secure=kwargs[‘secure’]
)
return cls.client

class MINio:
“””
大部分 minio api的方法。
具体使用方法可以参照官方文档 http://docs.minio.org.cn/docs/master/python-client-api-reference
也可使用 client 对象
“””

def __init__(self, *args, **kwargs):
self.minio_client = MINioClient(**kwargs)

@exc_handler
def put_object(self, file_path, bucket_name, object_name=None):
“””
添加一个新的对象到对象存储服务
:param file_path: 文件路径
:param bucket_name: 桶名称
:param object_name: 对象名称
:return:
“””
obj_name = object_name if object_name else uuid.uuid4().hex
file_size = os.stat(file_path).st_size
with open(file_path, ‘rb’) as file_data:
obj = self.minio_client.put_object(
bucket_name, obj_name, file_data, file_size,
)
return object_name, obj

@exc_handler
def fput_object(self, file_path, bucket_name, object_name=None):
“””
本地文件上传
:param file_path: 文件路径
:param bucket_name:
:param object_name:
:return:
“””
object_name = object_name if object_name else uuid.uuid4().hex
obj = self.minio_client.fput_object(bucket_name, object_name, file_path)
return object_name, obj

@exc_handler
def remove_object(self, bucket_name, object_name):
“””删除对象”””
self.minio_client.remove_object(bucket_name, object_name)

@exc_handler
def make_bucket(self, bucket_name, location=’us-east-1′):
“””
创建桶
:param bucket_name:
:param location:
:return: True success. False bucket already exists.
“””
if not self.minio_client.bucket_exists(bucket_name):
self.minio_client(bucket_name, location)
return True
return False

@exc_handler
def get_object(self, bucket_name, object_name, save_path: str = None, chunk_size: int = 32 * 1024):
“””
下载一个对象。 save_path 路径存在将保存到本地
:param bucket_name: 桶名称
:param object_name: 对象名称
:param save_path: 保存路径及文件名称
:param chunk_size: 每次读取文件大小
:return:
“””

data = self.minio_client.get_object(bucket_name, object_name)
if not save_path:
return data
write_file(save_path, data, chunk_size)

@exc_handler
def fget_object(self, bucket_name, object_name, file_path):
“””
下载文件到本地
:param bucket_name:
:param object_name:
:param file_path:
:return:
“””
self.minio_client(bucket_name, object_name, file_path)

@exc_handler
def list_objects(self, bucket_name, prefix=None, recursive=False):
“””列出存储桶所有对象，返回迭代器对象”””
return self.minio_client.list_objects(bucket_name, prefix, recursive)

@exc_handler
def list_buckets(self):
“””返回所有存储桶”””
return self.minio_client.list_buckets()