"""
我們的目的是借助單例模式，只生成乙個logger。
當日誌檔案已經存在時，預設以追加方式寫入。
"""import logging
import os
import time
import sys
import threading
defsynchronized
(func)
: func.__lock__ = threading.lock(
)def
lock_func
(*args,
**kwargs)
:with func.__lock__:
return func(
*args,
**kwargs)
return lock_func
class
singletontype
(type):
_instances =
@synchronized
def__call__
(cls,
*args,
**kwargs)
:if cls not
in cls._instances:
cls._instances[cls]
=super
(singletontype, cls)
.__call__(
*args,
**kwargs)
return cls._instances[cls]
class
logger
(metaclass=singletontype)
:def
__new__
(cls, log_file=
none
, file_output=
true
, standard_output=
true):
""" 1. logger的__new__只在建立第乙個例項的時候被呼叫。
2. 我們要返回的不是logger類例項，而是使用內建logging模組建立的例項。
"""# 日誌檔案的路徑
if log_file is
none
:# 檔案的當前目錄
current_path = os.path.dirname(os.path.realpath(__file__)
) current_time = time.strftime(
"%y-%m-%d-%h%m%s"
, time.localtime())
log_file = os.path.join(current_path,
'log_output'
, f'.txt'
) os.makedirs(os.path.dirname(log_file)
, exist_ok=
true
) cls.logger = logging.getlogger(__name__)
formater = logging.formatter(
'%(asctime)s %(name)s [%(filename)s %(lineno)s] %(message)s'
)# 如果既不使用標準輸出，也不使用日誌檔案輸出，則選擇標準輸出
ifnot file_output and
not standard_output:
standard_output =
true
if standard_output:
# 標準輸出
standard_out = logging.streamhandler(sys.stdout)
standard_out.setformatter(formater)
cls.logger.addhandler(standard_out)
# 新增標準輸出
if file_output:
# 日誌檔案輸出
file_out = logging.filehandler(log_file, encoding=
'utf-8'
) file_out.setformatter(formater)
cls.logger.addhandler(file_out)
# 新增檔案輸出
cls.logger.setlevel(logging.info)
return cls.logger
if __name__ ==
'__main__'
: logger = logger(
"logger1"
)# 只第乙個名稱有效
logger2 = logger(
"logger2"
) logger3 = logger(
"logger3"
) logger.info(
"this is a info"
)print
(logger is logger2, logger2 is logger3)
print
(not
false
andnot
false
)
"""
有n個執行緒，向同乙個信箱(信箱的大小為k)裡傳遞訊息，放入的訊息為數字i(i=1...n)，有m個執行緒從信箱裡取數字，取出來將數字*10並列印。
"""import time
import random
from threading import thread, semaphore, lock
from single import logger
list_len =
5in_index, out_index =0,
0in_lock, out_lock, output_lock = lock(
), lock(
), lock(
)message_list =[0
for x in
range
(list_len)
]full_semaphore, empty_semaphore = semaphore(0)
, semaphore(list_len)
defproducer
(name, num:
int)
:global in_index, in_lock, output_lock, list_len, message_list, full_semaphore, empty_semaphore
logger = logger(
) output_lock.acquire(
) logger.info(f' 初始化'
) output_lock.release(
)while
true
:# time.sleep(random.randint(1, 5))
empty_semaphore.acquire(
) in_lock.acquire(
) output_lock.acquire(
) message_list[in_index]
= num
in_index =
(in_index +1)
% list_len
logger.info(f'已經加入乙個: '
) output_lock.release(
) in_lock.release(
) full_semaphore.release(
)def
consumer
(name)
:global out_index, out_lock, output_lock, list_len, message_list, full_semaphore, empty_semaphore
logger = logger(
) output_lock.acquire(
) logger.info(f' 初始化'
) output_lock.release(
)while
true
:# time.sleep(random.randint(1, 5))
full_semaphore.acquire(
) out_lock.acquire(
) output_lock.acquire(
) num = message_list[out_index]*10
out_index =
(out_index +1)
% list_len
logger.info(f'已經取得乙個: '
) output_lock.release(
) out_lock.release(
) empty_semaphore.release(
)if __name__ ==
'__main__'
: _ =
for i in
range(1
,10):
p = thread(target=producer, args=
(f'producer- '
, i,))
p.start(
)for i in
range(1
,10):
c = thread(target=consumer, args=
(f'consumer- ',)
) c.start(
)for t in _:
t.join(
)
python 實現執行緒安全的單例模式
單例模式是一種常見的設計模式，該模式的主要目的是確保某乙個類只有乙個例項存在。當你希望在整個系統中，某個類只能出現乙個例項時，單例物件就能派上用場。比如，伺服器的配置資訊寫在乙個檔案中online.conf中，客戶端通過乙個 config 的類來讀取配置檔案的內容。如果在程式執行期間，有很多地方都需...
執行緒安全的單例模式
廢話不多說，常用的 積澱下來。一 懶漢模式 即第一次呼叫該類例項的時候才產生乙個新的該類例項，並在以後僅返回此例項。需要用鎖，來保證其執行緒安全性 原因 多個執行緒可能進入判斷是否已經存在例項的if語句，從而non thread safety.使用double check來保證thread safe...
執行緒安全的單例模式
單例模式是為了保證乙個類只有乙個例項而且易於外界訪問。所以一般只有把建構函式，拷貝函式，析構函式，賦值函式，變數名變為私有。再用乙個get函式訪問提供介面即可。考慮執行緒安全就要加鎖。一 懶漢模式 1 靜態成員例項的懶漢模式 class singleton public static singlet...