Docker API 与客户端库

Begindev 编辑部 April 1, 2026

在本教程中我们一直使用的 Docker 命令行界面（CLI）——从运行容器到构建镜像，再到管理网络——是一个极其强大的工具。然而，这只是与 Docker 交互的其中一种方式。在底层，Docker CLI 是通过 RESTful API 与 Docker Daemon 进行通信的。这个应用程序编程接口（API）是 Docker 的可编程接口，它允许你进行更细粒度的控制、自动化操作，以及与其他系统进行深度集成。

直接理解并使用 Docker API（通常通过客户端库，也被称为 SDK）将为你打开一个充满无限可能的新世界：你可以开发自定义工具、进行复杂的编排、构建自动化部署流水线，或者将 Docker 的能力直接集成到你自己的应用程序中。本章将为你揭开 Docker API 的神秘面纱，并向你展示如何利用客户端库通过代码来与你的 Docker 环境进行交互。

1. Docker Daemon API

在核心层面，Docker 采用的是客户端-服务器（Client-Server）模型。Docker Daemon（通常简称为“Docker 引擎”）是运行在你宿主机后台的服务器组件。它包揽了所有繁重的工作：构建镜像、运行容器、管理网络和持久化存储。而 Docker CLI 扮演的就是客户端的角色，负责向 Daemon 发送命令。

这种通信正是通过 Docker API 完成的。它是一个 RESTful API，这意味着它是基于标准的 HTTP 请求（GET、POST、PUT、DELETE）向特定的端点发送数据，并且通常以 JSON 格式返回结果。你执行的每一个 docker 命令都会被转换成一个或多个 API 调用。例如，当你输入 docker run hello-world 时，CLI 会将其转换为向 Docker Daemon 发送的 API 请求，要求其从 hello-world 镜像创建并启动一个新容器。

1.1 API 端点暴露方式

Docker Daemon 通过不同的机制来暴露它的 API：

Unix Socket (Linux/macOS)：在 Linux 和 macOS 上，Docker Daemon 通常通过位于 /var/run/docker.sock 的 Unix socket 暴露 API。这是在本地与 daemon 交互的默认且最安全的方式，因为对 socket 的访问受到文件系统权限的控制。通常只有 docker 用户组的成员（或 root 用户）才能访问它。
命名管道 (Named Pipe, Windows)：在 Windows 上，Docker Daemon 使用命名管道（通常是 \\.\pipe\docker_engine）进行本地通信。
TCP Socket： Docker Daemon 也可以被配置为监听 TCP 端口，通常是 2375（未加密）或 2376（TLS 加密）。这允许远程访问 Docker API，使你能够管理其他机器上的 Docker daemon。出于安全考虑，强烈不建议在没有 TLS 加密的情况下通过 TCP 暴露 Docker API，因为这会让任何能访问该端口的人完全控制你的 Docker 主机。

2. 为什么要直接与 API 交互？

虽然 Docker CLI 非常适合手动操作和编写简单的 Shell 脚本，但通过客户端库直接与 API 交互对于以下场景来说是必不可少的：

自动化： 构建自定义脚本或应用程序，根据特定的触发器或时间表自动扩缩容容器、部署服务或执行复杂的管理任务。
系统集成： 将 Docker 功能整合到现有的监控系统、CI/CD 流水线或专有平台中。例如，一个仪表盘应用可以通过查询各个 Docker 主机的 API 来显示跨主机的容器状态。
自定义工具： 开发超越 CLI 功能的专用工具，提供独特的用户界面，或者将 Docker 操作与其他系统功能结合起来。
动态工作流： 创建将动态配置、管理和销毁 Docker 资源作为其核心逻辑一部分的应用程序。

2.1 真实的直接交互案例

云服务提供商集成： 当你使用 AWS ECS、Google Kubernetes Engine (GKE) 或 Azure Container Instances (ACI) 等云服务时，这些服务通常使用底层容器运行时（可能是 Docker 或兼容的 containerd）的 API 来进行交互。例如，云环境中的自动扩缩容组可以监控 CPU 使用率，如果超过阈值，就会调用 API 在虚拟机上配置新的 Docker 容器。
CI/CD 流水线定制： 像 Jenkins 或 GitLab CI 这样的持续集成/持续部署服务器可能有插件或自定义脚本，在代码成功提交后，直接与 Docker API 交互来构建镜像、推送到注册中心或部署到测试环境。这允许实现通过简单 CLI 命令无法完成的高度定制化部署策略。

假设场景：智能家居系统

想象你正在构建一个智能家居自动化系统，其中各种微服务都运行在 Docker 容器中。这个系统需要：

只有当你不在家时，才启动“安全摄像头画面”容器。
当你睡觉时，停止“音乐服务器”容器以节省资源。
当有新版本可用时，动态更新“气象站”容器的镜像。你的自定义智能家居中枢应用可以使用 Docker 客户端库来检查你的状态，进行 API 调用来启动/停止特定容器，或者拉取和重建带有更新镜像的容器，而完全不需要手动输入 docker 命令。

3. Docker 客户端库 (Client Libraries)

虽然理论上你可以自己向 Docker API 发送原始 HTTP 请求，但使用客户端库要实用和高效得多。客户端库（或 SDK - 软件开发工具包）在原始 HTTP API 之上提供了一个特定于语言的抽象层，提供了一种更方便、更符合编程习惯的方式来与 Docker 交互。它们处理 HTTP 请求、JSON 解析、错误处理和身份验证的细节，让你专注于应用程序的业务逻辑。

许多编程语言都有官方或社区支持的 Docker 客户端库。常见的包括：

Python： docker-py（通常在代码中导入为 docker）
Go： github.com/docker/docker/client
Java： docker-java
Node.js： dockerode
Ruby： docker-api
PHP： docker-php

在本章中，我们将重点关注 Python 的 docker-py 库，因为它使用广泛、文档齐全，并且能非常有效地演示核心概念。

3.1 安装 docker-py

在使用该库之前，你需要先安装它：

pip install docker

4. 使用 docker-py 的实战案例

让我们深入实战。请确保你的 Docker Daemon 正在运行。

4.1 连接到 Docker Daemon

第一步始终是创建一个能够连接到你的 Docker Daemon 的客户端对象。docker-py 非常智能，它会尝试自动连接到默认的 Unix socket (/var/run/docker.sock) 或命名管道 (\\.\pipe\docker_engine)。

import docker

# 初始化 Docker 客户端
# 这将尝试连接到默认的 Docker socket
try:
    client = docker.from_env()
    print("成功连接到 Docker Daemon。")
except docker.errors.DockerException as e:
    print(f"连接 Docker Daemon 失败: {e}")
    print("请确保 Docker 正在运行并且可以访问。")
    exit(1)

# 你也可以连接到特定的 URL（例如，连接远程 Daemon）
# from docker import DockerClient
# client = DockerClient(base_url='tcp://192.168.1.100:2375') # 未加密的 TCP
# client = DockerClient(base_url='tcp://192.168.1.100:2376', tls=True) # TLS 加密的 TCP

4.2 列出容器（模拟 docker ps）

你可以列出所有运行中和已停止的容器，类似于 docker ps -a。

import docker

client = docker.from_env()

print("--- 列出所有容器（运行中和已停止） ---")
# 'all=True' 参数等同于 'docker ps -a'
containers = client.containers.list(all=True)

if not containers:
    print("没有找到任何容器。")
else:
    for container in containers:
        print(f"ID: {container.short_id}, 名称: {container.name}, 状态: {container.status}, 镜像: {container.image.tags}")

# 访问容器属性的方式：
# container.id (完整 ID)
# container.short_id (截断的短 ID)
# container.name (容器名称)
# container.status (容器状态)
# container.image.tags (镜像标签列表)
# container.attrs (包含所有原始 API 属性的字典)

4.3 运行新容器（模拟 docker run）

让我们运行一个简单的 hello-world 容器。

import docker
import time

client = docker.from_env()

print("--- 运行一个新的 'hello-world' 容器 ---")
try:
    # 运行 'hello-world' 镜像。
    # 'remove=True' 参数确保容器在退出后被自动删除。
    container = client.containers.run('hello-world', remove=True)
    print(f"容器 '{container.name}' 已启动并退出。")
    
    # 获取并打印容器的日志
    logs = container.logs().decode('utf-8')
    print("\n容器日志:")
    print(logs)
except docker.errors.ImageNotFound:
    print("本地未找到 'hello-world' 镜像。正在拉取...")
    client.images.pull('hello-world')
    print("镜像拉取完成。重试运行容器。")
    container = client.containers.run('hello-world', remove=True)
    logs = container.logs().decode('utf-8')
    print("\n容器日志:")
    print(logs)
except docker.errors.ContainerError as e:
    print(f"容器退出并报错: {e}")
except docker.errors.APIError as e:
    print(f"Docker API 错误: {e}")

print("\n--- 在后台模式运行 Nginx 容器 ---")
try:
    # 在后台模式运行 Nginx 容器 (-d)
    # 给它命名，将容器的 80 端口映射到主机的 8080 端口 (-p 8080:80)
    # 'detach=True' 参数类似于 -d
    # 'ports' 参数映射 host_port:container_port
    nginx_container = client.containers.run(
        'nginx:latest',
        name='my-nginx-webserver',
        ports={'80/tcp': 8080},
        detach=True,
        remove=False # 我们想让它保持运行以便后续交互
    )
    print(f"Nginx 容器 '{nginx_container.name}' 已启动 (ID: {nginx_container.short_id})。")
    print("请访问 http://localhost:8080")
    print("等待 5 秒钟以展示其正在运行...")
    time.sleep(5)
    print(f"Nginx 容器状态: {nginx_container.status}")
except docker.errors.ContainerError as e:
    print(f"Nginx 容器退出并报错: {e}")
except docker.errors.APIError as e:
    print(f"Docker API 错误: {e}")

# 清理 Nginx 容器
if 'nginx_container' in locals() and nginx_container.status == 'running':
    print(f"\n--- 停止并删除 Nginx 容器 '{nginx_container.name}' ---")
    nginx_container.stop()
    nginx_container.remove()
    print("Nginx 容器已停止并删除。")

4.4 管理容器生命周期（启动、停止、重启、强制停止、删除）

你可以通过 ID 或名称获取现有容器的引用，然后对它执行操作。

import docker
import time

client = docker.from_env()

# 确保有一个可以用于管理的测试容器
print("--- 确保存在一个用于管理的测试容器 ---")
try:
    test_container = client.containers.run('alpine', command='sleep 3600', name='my-test-container', detach=True)
    print(f"测试容器 '{test_container.name}' 已启动 (ID: {test_container.short_id})。状态: {test_container.status}")
except docker.errors.ContainerError as e:
    print(f"无法启动容器: {e}。尝试获取现有容器。")
    try:
        test_container = client.containers.get('my-test-container')
        print(f"找到现有容器 '{test_container.name}'。状态: {test_container.status}")
        if test_container.status == 'exited':
            print("容器已退出。正在启动它。")
            test_container.start()
            # 刷新容器对象以获取更新后的状态
            test_container.reload()
            print(f"容器已启动。新状态: {test_container.status}")
    except docker.errors.NotFound:
        print("无法启动或找到 'my-test-container'。退出程序。")
        exit(1)

# 停止容器
print(f"\n--- 停止容器 '{test_container.name}' ---")
test_container.stop()
test_container.reload() # 重新加载属性以获取最新状态
print(f"停止后的容器状态: {test_container.status}")
time.sleep(2)

# 启动容器
print(f"\n--- 启动容器 '{test_container.name}' ---")
test_container.start()
test_container.reload()
print(f"启动后的容器状态: {test_container.status}")
time.sleep(2)

# 重启容器
print(f"\n--- 重启容器 '{test_container.name}' ---")
test_container.restart()
test_container.reload()
print(f"重启后的容器状态: {test_container.status}")
time.sleep(2)

# 强制停止 (Kill) 容器
print(f"\n--- 强制停止容器 '{test_container.name}' ---")
test_container.kill()
test_container.reload()
print(f"Kill 后的容器状态: {test_container.status}")
time.sleep(2)

# 删除容器
print(f"\n--- 删除容器 '{test_container.name}' ---")
test_container.remove()
try:
    client.containers.get('my-test-container')
    print("容器仍然存在，删除失败。")
except docker.errors.NotFound:
    print("容器成功删除。")

4.5 列出和管理镜像（模拟 docker images, docker pull, docker rmi）

你可以通过代码交互式地管理 Docker 镜像。

import docker

client = docker.from_env()

print("--- 列出所有镜像 ---")
images = client.images.list()
if not images:
    print("没有找到镜像。")
else:
    for image in images:
        print(f"ID: {image.short_id}, 标签: {image.tags}")

print("\n--- 拉取镜像 (例如: 'ubuntu:latest') ---")
try:
    # 拉取镜像。这等同于 'docker pull ubuntu:latest'
    ubuntu_image = client.images.pull('ubuntu:latest')
    print(f"镜像 'ubuntu:latest' 已拉取。标签: {ubuntu_image.tags}")
except docker.errors.APIError as e:
    print(f"拉取镜像错误: {e}")

print("\n--- 删除镜像 (例如: 'ubuntu:latest') ---")
# 在尝试删除之前，确保该镜像未被任何容器使用
try:
    # 通过标签获取镜像对象
    image_to_remove = client.images.get('ubuntu:latest')
    # 删除镜像。等同于 'docker rmi ubuntu:latest'
    client.images.remove(image_to_remove.id)
    print("镜像 'ubuntu:latest' 成功删除。")
except docker.errors.ImageNotFound:
    print("未找到镜像 'ubuntu:latest'，跳过删除。")
except docker.errors.APIError as e:
    print(f"删除镜像错误: {e}")
    if "conflict" in str(e).lower():
        print("提示: 镜像可能正被容器使用。请先删除使用此镜像的容器。")

4.6 构建镜像（模拟 docker build）

对于 CI/CD 流水线来说，这就是展现其强大实力的地方。你可以从 Dockerfile 构建镜像。

首先，让我们创建一个简单的 Dockerfile 和应用程序文件。创建一个名为 my_app 的目录，包含以下文件：

my_app/Dockerfile:

# 用于简单 Python Flask 应用的 Dockerfile
FROM python:3.9-slim-buster
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["python", "app.py"]

my_app/requirements.txt:

Flask

my_app/app.py:

from flask import Flask
app = Flask(__name__)

@app.route('/')
def hello():
    return "Hello from Flask inside a Docker container!"

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

现在，使用 docker-py 构建并运行这个镜像。

import docker
import os
import time

client = docker.from_env()

# 定义包含 Dockerfile 的目录路径
dockerfile_path = './my_app' # 假设 my_app 目录与此脚本在同一位置
image_tag = 'my-flask-app:1.0'

if not os.path.exists(dockerfile_path):
    print(f"错误: 未找到目录 '{dockerfile_path}'。请创建它并在其中放入 Dockerfile, app.py 和 requirements.txt。")
    exit(1)

print(f"--- 正在从 '{dockerfile_path}' 构建镜像 '{image_tag}' ---")
try:
    # 构建镜像。'path' 是构建上下文，'tag' 是镜像标签。
    # 'build' 方法返回一个元组: (image_object, build_logs_generator)
    image, build_logs = client.images.build(path=dockerfile_path, tag=image_tag)
    
    print("构建日志:")
    for chunk in build_logs:
        if 'stream' in chunk:
            print(chunk['stream'], end='')
        if 'error' in chunk:
            print(f"构建过程中发生错误: {chunk['error']}")
            
    print(f"\n镜像 '{image_tag}' 构建成功 (ID: {image.short_id})。")
    
    print(f"\n--- 正在从镜像 '{image_tag}' 运行容器 ---")
    flask_container = client.containers.run(
        image_tag,
        name='my-flask-web-app',
        ports={'5000/tcp': 5000},
        detach=True
    )
    print(f"容器 '{flask_container.name}' 已启动 (ID: {flask_container.short_id})。请访问 http://localhost:5000")
    print("等待 5 秒钟让应用启动...")
    time.sleep(5)
    print(f"容器状态: {flask_container.status}")
    
    # 你可以检查容器以验证端口映射
    # print("\n容器审查信息:")
    # print(flask_container.attrs['NetworkSettings']['Ports'])

except docker.errors.BuildError as e:
    print(f"构建镜像出错: {e}")
    # 需要时你可以访问关于构建错误的详细信息
    for line in e.build_log:
        if 'stream' in line:
            print(line['stream'], end='')
except docker.errors.APIError as e:
    print(f"构建或运行期间发生 Docker API 错误: {e}")
finally:
    # 清理容器
    if 'flask_container' in locals():
        print(f"\n--- 停止并删除容器 '{flask_container.name}' ---")
        flask_container.stop()
        flask_container.remove()
        print("容器已停止并删除。")
    
    # 清理镜像
    try:
        print(f"--- 删除镜像 '{image_tag}' ---")
        client.images.remove(image_tag)
        print("镜像已删除。")
    except docker.errors.ImageNotFound:
        print(f"未找到镜像 '{image_tag}'，跳过删除。")
    except docker.errors.APIError as e:
        print(f"删除镜像 '{image_tag}' 出错: {e}")

4.7 获取容器日志（模拟 docker logs）

流式传输日志对于监控至关重要。

import docker
import time

client = docker.from_env()

print("--- 运行一个用于演示日志的容器 ---")
# 运行一个 alpine 容器，每秒打印一条消息
log_container = client.containers.run(
    'alpine/git', # 使用 alpine/git 因为它包含 'sh' 和 'sleep'
    command='sh -c "i=0; while true; do echo \'Hello from container: $i\'; i=$((i+1)); sleep 1; done"',
    name='log-demo-container',
    detach=True
)
print(f"容器 '{log_container.name}' 已启动。等待 3 秒钟积累日志。")
time.sleep(3)

print("\n--- 从容器流式传输日志 ---")
# 流式传输日志。'stream=True' 意味着它会在产生日志时逐行生成。
# 'follow=True' 保持数据流打开，就像 'tail -f' 一样。
# 'decode=True' 将字节解码为字符串。
for line in log_container.logs(stream=True, follow=False, decode=True):
    # 出于演示目的，只获取前几行内容
    if line.strip().startswith('Hello'):
        print(f"日志: {line.strip()}")
        # 你可以在这里添加逻辑来解析日志、触发警报等。
    
    # 为了演示 'follow=True'，你需要一个单独的线程或进程来停止容器。
    # 作为一个简单的示例，我们只获取当前日志而不一直跟随。

# 要获得一个可以停止的受控数据流：
# log_generator = log_container.logs(stream=True, follow=True, decode=True)
# try:
#     for _ in range(5): # 获取 5 行然后停止
#         line = next(log_generator)
#         print(f"流式日志: {line.strip()}")
# except StopIteration:
#     pass # 流结束，或者暂时没有更多日志

print(f"\n--- 停止并删除容器 '{log_container.name}' ---")
log_container.stop()
log_container.remove()
print("容器已停止并删除。")

4.8 审查 Docker 对象（容器、镜像、网络、卷）

inspect 功能提供了关于 Docker 对象的详细底层信息，类似于 docker inspect。这对于通过代码收集配置、运行时状态和网络详细信息非常有用。

import docker

client = docker.from_env()

# 确保存在一个容器用于审查
try:
    container = client.containers.get('my-nginx-webserver')
    if container.status == 'exited':
        container.start()
        container.reload()
except docker.errors.NotFound:
    print("未找到 Nginx 容器。正在运行一个用于审查...")
    container = client.containers.run('nginx:latest', name='my-nginx-webserver', ports={'80/tcp': 8080}, detach=True)
    print(f"Nginx 容器 '{container.name}' 已启动。")

print(f"\n--- 审查容器 '{container.name}' ---")
# .attrs 属性保存了 API inspect 调用返回的原始字典
container_info = container.attrs
print(f"容器名称: {container_info['Name']}")
print(f"容器状态: {container_info['State']['Status']}")
print(f"IP 地址: {container_info['NetworkSettings']['IPAddress']}")
print(f"80/tcp 的主机端口: {container_info['NetworkSettings']['Ports']['80/tcp'][0]['HostPort']}")

# 审查一个镜像
print("\n--- 审查镜像 'nginx:latest' ---")
try:
    image = client.images.get('nginx:latest')
    image_info = image.attrs
    print(f"镜像 ID: {image_info['Id']}")
    print(f"创建时间: {image_info['Created']}")
    print(f"镜像大小: {image_info['Size']} 字节")
    print(f"系统架构: {image_info['Architecture']}")
except docker.errors.ImageNotFound:
    print("未找到镜像 'nginx:latest'。先拉取它...")
    client.images.pull('nginx:latest')
    image = client.images.get('nginx:latest')
    image_info = image.attrs
    print(f"镜像 ID: {image_info['Id']}")
    print(f"创建时间: {image_info['Created']}")
    print(f"镜像大小: {image_info['Size']} 字节")
    print(f"系统架构: {image_info['Architecture']}")

# 清理工作
if 'container' in locals():
    print(f"\n--- 停止并删除容器 '{container.name}' ---")
    container.stop()
    container.remove()
    print("容器已停止并删除。")