container 示例源码

{
  "name": "container",
  "main": "index.js",
  "compatibility_date": "2026-01-01",
  "compatibility_flags": ["nodejs_compat"],
  "containers": [
    {
      "class_name": "MyContainer",
      "image": "./Dockerfile",
      "instance_type": "basic"
    }
  ],
  "durable_objects": {
    "bindings": [{ "name": "MY_CONTAINER", "class_name": "MyContainer" }]
  },
  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyContainer"] }]
}

{
  "name": "celld-container-example",
  "private": true,
  "type": "module",
  "dependencies": {
    "@cloudflare/containers": "^0.3.7"
  }
}

import { Container, getContainer, getRandom } from "@cloudflare/containers";

export class MyContainer extends Container {
  defaultPort = 8080;
  sleepAfter = "2m";
  envVars = { MESSAGE: "passed in by the Container class" };

  onStart() {
    console.log("container started");
  }
  onStop() {
    console.log("container stopped");
  }
  onError(error) {
    console.log("container error:", error);
  }
}

export default {
  async fetch(request, env) {
    const { pathname } = new URL(request.url);
    if (pathname.startsWith("/container/")) {
      // 每条路径对应一个容器，每个名称都有独立的对象和进程。
      return getContainer(env.MY_CONTAINER, pathname).fetch(request);
    }
    if (pathname.startsWith("/lb")) {
      // 将请求分散到三个容器。
      const container = await getRandom(env.MY_CONTAINER, 3);
      return container.fetch(request);
    }
    if (pathname.startsWith("/singleton")) {
      return getContainer(env.MY_CONTAINER).fetch(request);
    }
    return new Response(
      "Try /container/<name>, /lb, or /singleton\n",
      { headers: { "content-type": "text/plain" } },
    );
  },
};

FROM python:3.12-alpine
WORKDIR /app
COPY server.py .
ENV PORT=8080
EXPOSE 8080
CMD ["python3", "server.py"]

# 一个小型 HTTP 服务器：每次响应都返回容器主机名、请求路径，
# 以及对象传入的 MESSAGE 环境变量。
import json
import os
import signal
import socket
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer

# 该进程在容器中为 PID 1，会忽略没有注册处理函数的信号。
# 对象的 stop() 会发送 SIGTERM，因此收到该信号时应退出。
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({
            "hostname": socket.gethostname(),
            "path": self.path,
            "message": os.environ.get("MESSAGE", ""),
        }).encode()
        self.send_response(200)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


HTTPServer(("0.0.0.0", int(os.environ["PORT"])), Handler).serve_forever()