单例模式的目的是：

让程序一直保持一个连接，发送数据，不要一直建立新连接 和 销毁旧连接。

 

socket的单例模式：

import socket
from conf.settings import selenium_sk_ip, selenium_sk_port


class ConnectSelenium(object):

    __flag = None

    # 单例模式
    def __new__(cls, *args, **kwargs):
        if not cls.__flag:
            cls.__flag = super().__new__(cls)
        return cls.__flag

    def __init__(self, flag=False):
        if 'sk' not in self.__dict__ or flag:
            sk = socket.socket()
            sk.connect((selenium_sk_ip, int(selenium_sk_port)))
            self.sk = sk

    def get_sk(self):
        return self.sk


if __name__ == '__main__':
    ip = '127.0.0.1'
    port = 9001
    obj1 = ConnectSelenium()
    obj2 = ConnectSelenium()
    obj3 = ConnectSelenium()
    sk = obj2.get_sk()
    print(sk)
    # sk.send(b'123')
    # sk.send(b'456')
    sk.close()

 

mysql的单例模式：

import pymysql
from conf.settings import (mysql_host, mysql_port, mysql_user, mysql_pwd, connect_to_db, db_charset)


class ConnectMysql(object):

    __flag = None

    # 单例模式
    def __new__(cls, *args, **kwargs):
        if not cls.__flag:
            cls.__flag = super().__new__(cls)
        return cls.__flag

    def __init__(self):
        if 'cursor' not in self.__dict__:
            # print('实例')
            mysql_conn = pymysql.connect(host=mysql_host, user=mysql_user, passwd=mysql_pwd, port=mysql_port, db=connect_to_db, charset=db_charset)
            # 指定参数表示返回字典结果的数据
            cursor = mysql_conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.mysql_conn = mysql_conn
            self.cursor = cursor

    def get_cursor(self):
        return self.cursor

    def commit_data(self):
        self.mysql_conn.commit()


if __name__ == '__main__':
    obj1 = ConnectMysql()
    obj2 = ConnectMysql()
    cursor = obj1.get_cursor()

 

redis的单例模式：

from redis import Redis
from conf.settings import (
    redis_host,
    redis_port,
    redis_db,
    redis_password
)


class ConnectRedis(object):

    __flag = None

    # 单例模式
    def __new__(cls, *args, **kwargs):
        if not cls.__flag:
            cls.__flag = super().__new__(cls)
        return cls.__flag

    def __init__(self):
        if 'cursor' not in self.__dict__:
            redis_conn = Redis(host=redis_host, port=redis_port, db=redis_db, password=redis_password)
            self.cursor = redis_conn

    def get_cursor(self):
        return self.cursor


if __name__ == '__main__':
    obj1 = ConnectRedis()
    obj2 = ConnectRedis()
    cursor = obj1.get_cursor()
    print(obj1)
    print(obj2)

 

日志的单例模式：

import logging


log_path = r'D:\log\t.txt'


# 单例模式
class Log(object):

    __flag = None

    def __new__(cls, *args, **kwargs):
        if not cls.__flag:
            cls.__flag = super().__new__(cls)
        return cls.__flag

    def __init__(self):
        if 'logger' not in self.__dict__:
            logger = logging.getLogger()
            logger.setLevel(level=logging.DEBUG)
            filehandle = logging.FileHandler(log_path,encoding='utf-8')
            streamhandle = logging.StreamHandler()
            logger.addHandler(filehandle)
            logger.addHandler(streamhandle)
            format = logging.Formatter('%(asctime)s:%(levelname)s:%(lineno)s %(message)s')
            filehandle.setFormatter(format)
            streamhandle.setFormatter(format)

            self.logger = logger

    def return_logger(self):
        return self.logger

def get_logger():
    return Log().return_logger()

if __name__ == '__main__':
    logger = get_logger()
    logger.debug('zezhou')