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()