大家一般都用logging日志打印，但logging是线程安全的，多进程也有很多介绍，引入一些文件锁，对logging做好配置，能过支持。
但通过测试，发现多进程时还是容易出现重复写入文件或者打印正常漏写入文件的问题。
我的日志需求比较简单，能够区分文件，正确的写入日志文件。
引入文件锁；日志写入函数封装到一个操作_logger类中； 日志名称和写入级别封装到一个业务类logger中。
本范例基于python3实现。本范例20个进程并发，分别写入3个文件，每s每个文件写入超过100行数据，日志文件中没有数据冗余，也没有数据遗漏。
# -*-coding:utf-8-*-
“””
author：yinshunyao
date:2017/3/5 0005下午 10:50
“””
# import logging
import os
import time
# 利用第三方系统锁实现文件锁定和解锁
if os.name == ‘nt’:
import win32con, win32file, pywintypes
lock_ex = win32con.lockfile_exclusive_lock
lock_sh = 0 # the default value
lock_nb = win32con.lockfile_fail_immediately
__overlapped = pywintypes.overlapped()
def lock(file, flags):
hfile = win32file._get_osfhandle(file.fileno())
win32file.lockfileex(hfile, flags, 0, 0xffff0000, __overlapped)
def unlock(file):
hfile = win32file._get_osfhandle(file.fileno())
win32file.unlockfileex(hfile, 0, 0xffff0000, __overlapped)
elif os.name == ‘posix’:
from fcntl import lock_ex
def lock(file, flags):
fcntl.flock(file.fileno(), flags)
def unlock(file):
fcntl.flock(file.fileno(), fcntl.lock_un)
else:
raise runtimeerror(“file locker only support nt and posix platforms!”)
class _logger:
file_path = ” #初始化日志路径
@staticmethod
def init():
if not _logger.file_path:
_logger.file_path = ‘%s/log’ % os.path.abspath(os.path.dirname(__file__))
return true
@staticmethod
def _write(messge, file_name):
if not messge:
return true
messge = messge.replace(‘\t’, ‘,’)
file = ‘{}/{}’.format(_logger.file_path, file_name)
while true:
try:
f = open(file, ‘a+’)
lock(f, lock_ex)
break
except:
time.sleep(0.01)
continue
# 确保缓冲区内容写入到文件
while true:
try:
f.write(messge + ‘\n’)
f.flush()
break
except:
time.sleep(0.01)
continue
while true:
try:
unlock(f)
f.close()
return true
except:
time.sleep(0.01)
continue
@staticmethod
def write(message, file_name, only_print=false):
if not _logger.init(): return
print(message)
if not only_print:
_logger._write(message, file_name)
class logger:
def __init__(self, logger_name, file_name=”):
self.logger_name = logger_name
self.file_name = file_name # 根据消息级别，自定义格式，生成消息
def _build_message(self, message, level):
try:
return ‘[%s]\t[%5s]\t[%8s]\t%s’ \
% (time.strftime(‘%y-%m-%d %h:%m:%s’), level, self.logger_name, message)
except exception as e:
print(‘解析日志消息异常：{}’.format(e))
return ”
def warning(self, message):
_logger.write(self._build_message(message, ‘warn’), self.file_name)
def warn(self, message):
_logger.write(self._build_message(message, ‘warn’), self.file_name)
def error(self, message):
_logger.write(self._build_message(message, ‘error’), self.file_name)
def info(self, message):
_logger.write(self._build_message(message, ‘info’), self.file_name, true)
def debug(self, message):
_logger.write(self._build_message(message, ‘debug’), self.file_name)
# 循环打印日志测试函数
def _print_test(count):
logger = logger(logger_name=’test{}’.format(count), file_name=’test{}’.format(count % 3))
key = 0
while true:
key += 1
# print(‘{}-{}’.format(logger, key))
logger.debug(‘%d’ % key)
logger.error(‘%d’ % key)
if __name__ == ‘__main__’:
from multiprocessing import pool, freeze_support
freeze_support() # 进程池进行测试
pool = pool(processes=20)
count = 0
while count < 20:
count += 1
pool.apply_async(func=_print_test, args=(count,))
else:
pool.close()
pool.join()
以上就是详解python日志打印和写入并发实现代码的详细内容，更多请关注 第一php社区 其它相关文章！