Chemmy's Blog

chengming0916@outlook.com

发表于 更新于

在Docker容器中部署运行Python应用时，日志打印是排查故障、监控服务运行状态的核心环节。与本地运行环境不同，Docker容器的文件系统具有临时性，且日志需被容器引擎统一捕获、管理。因此，Python在Docker环境下打印日志的核心原则是：将日志直接输出至标准输出（stdout）和标准错误（stderr），由Docker自动捕获，无需手动写入本地文件。本文汇总了实用实现方案、常见问题解决方案及生产环境最佳实践，帮助开发者高效、规范地实现Docker环境下的Python日志打印。

一、基础日志打印方式：print函数（快速上手）

对于简单测试场景或小型脚本开发，Python内置的print()函数是最便捷的日志打印方式。print()函数默认将内容输出至stdout，而Docker会自动捕获stdout与stderr的输出内容，无需额外配置即可快速查看日志。

示例代码：

1
2
3
print("Python应用启动中...")
print("核心功能执行完成")
print("错误：参数传入异常", file=sys.stderr)  # 输出到stderr，区分正常日志与错误日志

Docker端核心日志查看命令：

1
2
3
4
5
# 查看指定容器的全部日志
docker logs 容器名/容器ID

# 实时跟踪日志（类似tail -f，适合监控服务状态）
docker logs -f 容器名/容器ID

注意：print()仅适用于简单场景，其不支持日志级别、时间戳等关键信息，无法满足生产环境下日志分类、追溯、分析的管理需求。

二、生产环境最佳实践：logging模块

对于正式项目或生产环境，推荐使用Python标准库中的logging模块。该模块支持日志级别（DEBUG、INFO、ERROR等）、自定义日志格式、时间戳等实用功能，且能完美适配Docker环境，核心配置要点是将日志强制输出至stdout，确保Docker能及时、准确捕获日志。

2.1 logging模块基础配置（Docker友好版）

配置时需明确指定日志输出至stdout，避免日志被缓冲或输出至其他位置，导致Docker无法捕获，具体示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import logging
import sys

# 配置日志：指定级别、格式、输出目标，适配Docker环境
logging.basicConfig(
    level=logging.INFO,  # 日志级别，可根据需求调整为DEBUG（调试）/ERROR（错误）
    format="%(asctime)s - %(levelname)s - %(message)s",  # 日志格式（包含时间戳、日志级别、日志内容）
    handlers=[
        logging.StreamHandler(sys.stdout)  # 强制输出到stdout，确保Docker正常捕获
    ]
)

# 初始化日志对象
logger = logging.getLogger(__name__)

# 打印不同级别的日志（适配不同场景的日志记录需求）
logger.debug("调试信息：参数初始化完成")
logger.info("服务启动成功，监听端口8000")
logger.warning("警告：配置文件未指定超时时间，建议补充")
logger.error("错误：数据库连接失败，请检查连接配置")

2.2 配置核心说明

  1. logging.StreamHandler(sys.stdout)：明确将日志输出至stdout，避免因Python默认输出方式差异，导致Docker日志捕获延迟、丢失或异常；

  2. 日志级别：需根据项目实际需求设置，生产环境建议使用INFO及以上级别，避免调试日志过多占用资源、干扰问题定位；

  3. 日志格式：包含时间戳（asctime）、日志级别（levelname）、日志内容（message），可快速追溯日志产生时间、类型及具体信息，便于后续日志分析和故障定位。

三、关键问题解决：Docker日志不输出/延迟

很多开发者在实践中会遇到“Python代码正常打印日志，但Docker logs无法查看”或“日志延迟输出”的问题，核心原因是Python默认开启输出缓冲，导致日志未及时刷新至stdout，Docker无法捕获。以下提供两种高效解决方案，优先推荐第二种，更适配生产环境。

3.1 运行时添加-u参数（快速临时解决）

在运行Python脚本时，添加-u参数（即unbuffered无缓冲模式），可强制日志实时输出，避免缓冲导致的日志卡住、延迟问题。

直接运行命令：

1
python -u app.py

Dockerfile中CMD指令写法：

1
CMD ["python", "-u", "app.py"]

3.2 设置环境变量（优雅持久解决）

在Dockerfile中添加PYTHONUNBUFFERED环境变量，其效果与-u参数完全一致，无需修改Python脚本运行命令，更适合生产环境的Docker镜像标准化构建。

Dockerfile中添加指令：

1
ENV PYTHONUNBUFFERED=1

该环境变量会直接关闭Python的输出缓冲机制，确保日志实时刷新至stdout，保障Docker能及时捕获每一条日志，避免延迟或丢失。

四、完整Dockerfile示例（可直接复用）

结合上述最佳实践，以下提供完整的Dockerfile示例，包含基础镜像选择、环境变量配置、代码复制及应用运行指令，可直接复用，确保Python日志正常打印、Docker可正常捕获。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 选择轻量Python基础镜像（推荐3.10+版本，兼顾性能与兼容性）
FROM python:3.11-slim

# 关键配置：关闭Python输出缓冲，确保日志实时输出至stdout
ENV PYTHONUNBUFFERED=1

# 设置容器工作目录，规范文件组织结构
WORKDIR /app

# 复制本地Python代码至容器工作目录
COPY . .

# 运行Python应用（无需额外添加-u参数，环境变量已生效）
CMD ["python", "app.py"]

五、Docker日志常用操作命令

日志打印后，需通过Docker命令查看、筛选、监控日志，以下汇总最常用的操作命令，覆盖日常开发、测试及运维场景，简单易上手。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 1. 查看指定容器的全部日志
docker logs 容器名/容器ID

# 2. 实时跟踪日志（实时监控服务运行状态，排查实时故障）
docker logs -f 容器名/容器ID

# 3. 查看最近N行日志（避免日志过多刷屏，快速定位最新问题）
docker logs --tail 100 容器名/容器ID  # 示例：查看最近100行日志

# 4. 带时间戳查看日志（精准定位具体时间点的故障日志）
docker logs -t 容器名/容器ID

# 5. 查看指定时间段的日志（需结合-t参数，精准筛选目标日志）
docker logs -t --since="2024-05-01T10:00:00" 容器名/容器ID

六、不推荐做法及避坑提醒

在Docker环境下，以下日志打印方式不推荐使用，易导致日志丢失、无法管理、排查困难等问题，需重点规避：

  1. 不推荐将日志写入容器本地文件：Docker容器的文件系统为临时存储，容器重启后日志会直接丢失；且需额外进入容器查看日志，操作繁琐、效率极低，不利于日志统一管理。

  2. 不推荐不配置logging.handlers：若未指定StreamHandler(sys.stdout)，部分Python版本会将日志默认输出至stderr或其他位置，可能导致Docker捕获日志异常、日志错乱。

  3. 不推荐忽略缓冲问题：未关闭Python输出缓冲，会导致日志延迟输出，排查故障时无法及时看到最新日志，增加问题定位难度。

七、总结

Python在Docker环境下打印日志的核心要点的是“标准化输出+无缓冲”，结合实践可总结为4个关键步骤，简单易落地：

  1. 简单测试场景：使用print()函数，直接输出至stdout/stderr，快速实现日志打印；

  2. 生产环境：使用logging模块，配置StreamHandler(sys.stdout)，自定义日志格式，满足日志分类、追溯需求；

  3. 避免日志延迟/丢失：通过设置ENV PYTHONUNBUFFERED=1或运行时添加-u参数，关闭Python输出缓冲；

  4. 日志管理：使用Docker日志命令（docker logs）查看、监控日志，无需手动管理日志文件，提升效率。

按照以上最佳实践，可确保Python应用在Docker环境下的日志清晰、实时、可管理，为后续故障排查、服务监控提供有力支撑，提升开发与运维效率。

发表于 更新于 分类于

方式一

通过udev规则监听设备事件，编写/etc/udev/rules.d/99-udev-mount.rules规则实现U盘插入捕获U盘插入事件

1
2
3
4
5
# 插入U盘自动挂载
ACTION=="add", KERNEL=="sd[a-z]*", RUN+="/bin/mkdir -p /media/udev-%k", RUN+="/bin/mount /dev/%k /media/udev-%k"

# 移除U盘自动卸载
ACTION=="remove", KERNEL=="sd[a-z]*", RUN+="/bin/umount /media/udev-%k"

规则编辑完成后执行以下命令使规则生效

1
sudo udevadm control --reload

发表于 更新于 分类于 ，

1. 准备环境

  • 准备开发包：包含头文件(.h)、库文件（.dll或.so）及对接文档
  • 安装依赖：确保Python环境已安装ctypes库或第三方库例如Cython(用于复杂场景)
  • 配置路径：将SDK的库路径添加到环境变量或直接在代码中指定路径(推荐方式，不会因为换电脑导致无法编译，例如sdk/windows/sdk.dll)

2. 封装接口

加载SDK

1
2
3
4
5
6
7
8
9
10
import sys
from ctypes import *
from ctypes import wintypes

# 区分Windows和Linux环境，加载不同SDK
if sys.platform.startwith("win"):
	sdk = WinDLL("sdk/windows/sdk.dll")
elif sys.platform.startwith("linux"):
	sdk = CDLL("sdk/linux/sdk.so")

定义结构体

1
2
3
4
5
6
7
# 定义结构体，需要与SDK头文件一致
class DEMO：
	_fields_ = [
		("fieldname-1", c_int),   # int 类型
		("fieldname-2", c_int_p), # int 指针
		# 其他字段参考SDK文档
	]

![[Python对接C库/IMG-20250804110742707.png]]
定义函数原型，需严格对齐SDK中的数据类型和函数参数顺序

1
2
3
4
sdk.Init.restype = c_bool # 映射返回值，Init为C/C++中的函数名
sdk.Init.argtypes = [     # 映射参数列表
	c_int, c_int_p, c_char_p
]

3. 接口调用

函数调用

1
sdk.Init(c_int(0), c_int_p(0), c_char_p(b"this is a test"))

带有回调函数的函数调用
回调函数例如

1
int (*Callback) (int, char*);

Python中定义回调函数类型

1
CallbackType = CFUNCTYPE(c_int, c_int, c_char_p) # 返回类型在前，参数在后

若C函数使用__stdcall（常见于Windows API）,需要WINFUNCTYPE替代CFUNCTYPE，若为__cdecl（默认）,则使用CFUNCTYPE
Python实现回调函数（参数和返回值需与C定义严格一致）

1
2
3
def py_callback(num, text) -> int:
	print(f"Received： {num}, {text.decode("utf-8")}")
	return 0 # 返回值需与C定义匹配

处理指针参数

若回调参数包含指针，例如void*，需要使用c_void_p类型，并通过cast解析

1
2
3
def py_callback(data_ptr): 
	data = cast(data_ptr, POINTER(c_int)).contents.value 
	return data

注册回调函数

1
2
3
4
c_callback = CallbackType(py_callback) # 使用定义的回调类型包装Python函数
global_keep_alive = c_callback # 关键！ 将回调对象保存为全局变量或类属性，防止被回收
sdk.register_callback.argtypes = [c_int, CallbackType]
sdk.register_callback.restype = None

4. 资源释放

退出时需要调用SDK中的清理函数释放资源

1
sdk.Cleanup()

5. 注意事项

  • 结构体指针和缓冲区需要手动分配/释放，避免内存泄漏
  • 不同版本SDK接口可能有差异，建议统一开发与部署环境
  • 映射Windows中特有的类型例如WORD,DWORD在wintypes包中
  • C调用Python回调时，若Python函数抛出异常可能导致程序崩溃。需要在回调内部处理异常。
  • 若C函数在子线程中调用回调，需确保Python的GIL（全局解释锁）已获取
    1
    2
    3
    4
    5
    6
    7
    from ctypes import py_object, pythonapi 
    PyGILState_Ensure = pythonapi.PyGILState_Ensure 
    PyGILState_Release = pythonapi.PyGILState_Release 
    def thread_safe_callback(): 
    	state = PyGILState_Ensure() 
    	# 执行Python操作 
    	PyGILState_Release(state)

发表于 更新于 分类于 ，

使用os.stat()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
package main
import (
   "fmt"
   "os"
)

func main() {
   dir := "new"
   if _, err := os.Stat(dir); os.IsNotExist(err) {
      fmt.Println(dir, "does not exist")
   } else {
      fmt.Println("The provided directory named", dir, "exists")
   }
}

使用os.open()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
package main
import (
   "fmt"
   "os"
)

func main() {
   dir := "go"
   if _, err := os.Open(dir); os.IsNotExist(err) {
      fmt.Println("The directory named", dir, "does not exist")
   } else {
      fmt.Println("The directory namend", dir, "exists")
   }
}

使用mkdir()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
package main
import (
   "fmt"
   "os"
)

func main() {
   dir := "new"
   if err := os.Mkdir(dir, 0755); os.IsExist(err) {
      fmt.Println("The directory named", dir, "exists")
   } else {
      fmt.Println("The directory named", dir, "does not exist")
   }
}

发表于 更新于 分类于 ，

ECB模式，PKCS5填充

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
package sm4
 
import (
	"bytes"
	"encoding/hex"
	"github.com/tjfoc/gmsm/sm4"
)
 
func SM4EcbEncrypt(key, plaintext []byte) (string, error) {
	block, err := sm4.NewCipher(key)
	if err != nil {
		return "", err
	}
 
	plaintext = PKCS5Padding(plaintext, block.BlockSize())
	ciphertext := make([]byte, len(plaintext))
 
	for start := 0; start < len(plaintext); start += block.BlockSize() {
		block.Encrypt(ciphertext[start:start+block.BlockSize()], plaintext[start:start+block.BlockSize()])
	}
 
	return hex.EncodeToString(ciphertext), nil
}
 
func SM4EcbDecrypt(key []byte, data string) ([]byte, error) {
	plaintext, _ := hex.DecodeString(data)
	block, err := sm4.NewCipher(key)
	if err != nil {
		return nil, err
	}
 
	ciphertext := make([]byte, len(plaintext))
 
	for start := 0; start < len(plaintext); start += block.BlockSize() {
		block.Decrypt(ciphertext[start:start+block.BlockSize()], plaintext[start:start+block.BlockSize()])
	}
	ciphertext = PKCS5Unpadding(ciphertext)
	return ciphertext, nil
}
 
func PKCS5Padding(ciphertext []byte, blockSize int) []byte {
	padding := blockSize - len(ciphertext)%blockSize
	padtext := bytes.Repeat([]byte{byte(padding)}, padding)
	return append(ciphertext, padtext...)
}
 
func PKCS5Unpadding(origData []byte) []byte {
	length := len(origData)
	unpadding := int(origData[length-1])
	return origData[:(length - unpadding)]
}

发表于 更新于 分类于 ，

使用sys模块

1
2
3
4
5
6
7
8
9
10
import sys

if sys.platform.startwith("win"):
	print("当前系统是Windows")
elif sys.platform.startwith("linux"):
	print("当前系统是Linux")
elif sys.platform.startwith("darwin"):
	print("当前系统是MAC OS")
else:
	print("当前系统是其他操作系统")

使用platform模块

1
2
3
4
5
6
7
8
9
10
11
import platform

system=platform.system()
if system=="Windows":
	print("当前系统是Windows")
elif system=="Linux":
	print("当前系统是Linux")
elif system=="Darwin":
	print("当前系统是MAC OS")
else:
	print("当前系统是其他操作系统")

使用os模块

1
2
3
4
5
6
7
8
9
import os

system = os.name
if system == "nt":
	print("当前系统是Windows")
elif system == "posix":
	print("当前系统是Linux或Mac OS")
else
	print("当前系统是其他操作系统")
0%