1. 获取第三方模块使用的日志实例名

只需要添加如下两行：

import logging logging.basicConfig(level=logging.DEBUG)

就能查看到第三方模块使用的日志实例名

比如我想使用 aiostomp 来异步操作 ActiveMQ 消息队列，使用 Docker 快速启动一个 ActiveMQ：

# docker run -d \ --name activemq \ -p 61613:61613 -p 8161:8161 \ --rm \ rmohr/activemq

阿里云镜像加速器： https://cr.console.aliyun.com/cn-hangzhou/instances/mirrors

消息的生产者代码：

# -*- coding: utf-8 -*- import asyncio from datetime import datetime import logging import time from aiostomp import AioStomp logging.basicConfig(level=logging.DEBUG) async def run(): client = AioStomp('192.168.80.2', 61613, error_handler=report_error) await client.connect() # 连接 ActiveMQ # 定时发送消息到队列 while True: destination = '/queue/channel' data = datetime.now().strftime('%Y-%m-%d %H:%M:%S') client.send(destination, body=data, headers={}) logging.info('Send a message({0}) to queue({1})'.format(data, destination)) time.sleep(1) async def report_error(error): logging.error('Catch error: ', error) if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(run()) # loop.run_forever()

运行它后控制台会输出如下日志：

DEBUG:asyncio:Using selector: EpollSelector DEBUG:aiostomp:connect INFO:aiostomp:Connecting to stomp server: 192.168.80.2:61613 INFO:aiostomp:Connected INFO:root:Send a message(2019-11-16 16:09:58) to queue(/queue/channel) INFO:root:Send a message(2019-11-16 16:09:59) to queue(/queue/channel) INFO:root:Send a message(2019-11-16 16:10:00) to queue(/queue/channel)

说明： asyncio 模块内部使用的日志实例名叫 asyncio，aiostomp 模块内部使用的日志实例名叫 aiostomp

2. 设置日志格式和颜色

安装 coloredlogs：

(venv)# pip install coloredlogs

创建 logger.py：

import logging import logging.handlers import os import coloredlogs # 获取默认 Logger 实例 root_logger = logging.getLogger() # 创建应用程序中使用的实例 logger = logging.getLogger('App') # 设置日志格式 fmt = '%(asctime)s - [%(name)+24s] - %(filename)+18s[line:%(lineno)5d] - %(levelname)+8s: %(message)s' formatter = logging.Formatter(fmt) # 创建输出到控制台的 Handler console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) # 创建输出到日志文件的 Handler，按时间周期滚动 basedir = os.path.abspath(os.path.dirname(__file__)) log_dest = os.path.join(basedir, 'logs') # 日志文件所在目录 if not os.path.isdir(log_dest): os.mkdir(log_dest) filename = os.path.join(log_dest, 'app.log') file_handler = logging.handlers.TimedRotatingFileHandler(filename, when='midnight', interval=1, backupCount=7, encoding='utf-8') file_handler.setFormatter(formatter) # 为实例添加 Handler root_logger.addHandler(console_handler) root_logger.addHandler(file_handler) # 设置日志级别 root_logger.setLevel(logging.DEBUG) # 当日志输出到控制台时，会带有颜色 coloredlogs.DEFAULT_FIELD_STYLES = dict( asctime=dict(color='green'), name=dict(color='blue'), filename=dict(color='magenta'), lineno=dict(color='cyan'), levelname=dict(color='black', bold=True), ) coloredlogs.install(fmt=fmt, level='DEBUG', logger=root_logger)

然后在生产者代码中引入：

# -*- coding: utf-8 -*- import asyncio from datetime import datetime import time from aiostomp import AioStomp from logger import logger async def run(): # 测试不同日志级别的颜色 logger.debug('This is debug message') logger.info('This is info message') logger.warning('This is warning message') logger.error('This is error message') logger.critical('This is critical message') # 连接 ActiveMQ client = AioStomp('192.168.80.2', 61613, error_handler=report_error) await client.connect() # 定时发送消息到队列 while True: destination = '/queue/channel' data = datetime.now().strftime('%Y-%m-%d %H:%M:%S') client.send(destination, body=data, headers={}) logger.info('Send a message({0}) to queue({1})'.format(data, destination)) time.sleep(1) async def report_error(error): logger.error('Catch error: ', error) if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(run()) # loop.run_forever()