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：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @File   : logger.py
# @Author : Madman
# @Date   : 2019/11/16 15:21
import logging
import os
import time

import coloredlogs


#-----------------------------------------------------------------------------
# 创建或获取 Logger 实例
#-----------------------------------------------------------------------------

logger = logging.getLogger(__name__)
# asyncio 模块使用的日志实例
asyncio_logger = logging.getLogger('asyncio')
# aiostomp 模块使用的日志实例
aiostomp_logger = logging.getLogger('aiostomp')


#-----------------------------------------------------------------------------
# 设置日志格式
#-----------------------------------------------------------------------------

fmt = '%(asctime)s - [%(name)s] - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s'
formatter = logging.Formatter(fmt)


#-----------------------------------------------------------------------------
# 创建 Handler, 输出日志到控制台和文件
#-----------------------------------------------------------------------------

# 日志文件 FileHandler
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 = time.strftime('%Y-%m-%d-%H-%M-%S', time.localtime(time.time())) + '.log'  # 日志文件名，以当前时间命名
file_handler = logging.FileHandler(os.path.join(log_dest, filename), encoding='utf-8')  # 创建日志文件handler
file_handler.setFormatter(formatter)  # 设置Formatter

# 控制台日志 StreamHandler
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)


#-----------------------------------------------------------------------------
# 为 Logger 添加 Handler
#-----------------------------------------------------------------------------

logger.addHandler(file_handler)
logger.addHandler(stream_handler)
asyncio_logger.addHandler(file_handler)
asyncio_logger.addHandler(stream_handler)
aiostomp_logger.addHandler(file_handler)
aiostomp_logger.addHandler(stream_handler)


#-----------------------------------------------------------------------------
# 设置日志级别
#-----------------------------------------------------------------------------

logger.setLevel(logging.DEBUG)
asyncio_logger.setLevel(logging.DEBUG)
aiostomp_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=logger)
coloredlogs.install(fmt=fmt, level='DEBUG', logger=asyncio_logger)
coloredlogs.install(fmt=fmt, level='DEBUG', logger=aiostomp_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()