进阶篇：VPC 对等连接与高级特性
2026/6/13...大约 14 分钟
0如果喜欢就点个赞叭0
运行前提条件
- 系统：Linux 系统（Ubuntu 20.04+ / Debian 11+ / CentOS 8+ 均可），Windows 用户用 WSL2 也可以
- 权限：必须有 sudo /root 权限（操作内核网络模块需要）
- 环境：Python 3.8+，不需要装任何第三方库，纯标准库就能跑
- 基础：懂一点 Linux 基础命令和 Python 语法就行，零基础跟着步骤也能跑通
一、学习目标
- 理解 VPC 对等连接的工作原理和使用场景
- 实现两个 VPC 之间的私有网络互通
- 掌握自定义路由和高级流量控制
- 了解企业级 VPC 架构设计原则
二、核心原理
VPC 对等连接是两个 VPC 之间的私有网络连接，不经过公网，具有以下优势：
- 安全：流量在私有网络内传输，不暴露在公网上
- 高速：低延迟、高带宽，和 VPC 内部通信速度相同
- 免费：大多数云厂商的对等连接都是免费的
它的实现原理非常简单：在两个 VPC 的路由表中添加对方的网段路由，让两个 VPC 的流量可以互相转发。
三、代码分段解析
我们将在第三篇互联 VPC 的基础上，添加 VPC 对等连接功能。
3.1 全局资源管理与基础类（聚焦版）
说明：为了聚焦对等连接的核心原理并保持代码简洁，我们在本篇的完整代码中去除了安全组、网络 ACL、NAT 网关等之前的进阶功能。VPC、Subnet、VM 的基础实现与第一篇一致，此处省略，完整代码见文末汇总。
3.2 VPC 对等连接类实现
做什么：模拟云厂商的 VPC 对等连接功能，实现两个 VPC 之间的私有互通。
为什么这么做：企业通常会创建多个 VPC 来隔离不同的环境（生产、开发、测试），但有时候需要这些环境之间互相通信。
class VPCPeering:
"""
模拟VPC对等连接：实现两个VPC之间的私有网络互通
对应云厂商控制台：VPC对等连接管理
"""
def __init__(self, name: str, vpc1: 'VPC', vpc2: 'VPC'):
self.name = name
self.vpc1 = vpc1
self.vpc2 = vpc2
# 添加互相的路由
vpc1.add_route(str(vpc2.cidr), vpc1.bridge_name)
vpc2.add_route(str(vpc1.cidr), vpc2.bridge_name)
# 允许两个VPC之间的流量转发
subprocess.run(
f"iptables -I FORWARD -s {vpc1.cidr} -d {vpc2.cidr} -j ACCEPT",
shell=True, check=True
)
subprocess.run(
f"iptables -I FORWARD -s {vpc2.cidr} -d {vpc1.cidr} -j ACCEPT",
shell=True, check=True
)
print(f"✅ 创建VPC对等连接 [{name}]：{vpc1.name} ↔ {vpc2.name}")关键要点：
- 对等连接是双向的，需要在两个 VPC 的路由表中都添加对方的路由
- 需要在 iptables 的 FORWARD 链中添加规则，允许两个 VPC 之间的流量转发
- 两个 VPC 的网段不能重叠，否则路由会冲突
3.3 交互式 CLI Shell（对等连接版）
做什么：通过命令行操作两个隔离的 VPC，建立对等连接，并验证连通性变化。
import cmd
import shlex
class VPCShell(cmd.Cmd):
intro = '===========================================================\n' \
'欢迎使用 VPC 交互式模拟器（对等连接篇）\n' \
'【可用指令】:\n' \
' 1. create_vpc <name> <cidr> - 创建 VPC\n' \
' 2. create_subnet <vpc_name> <name> <cidr> - 创建子网\n' \
' 3. create_vm <subnet_cidr> <vm_name> [ip] - 创建虚拟机\n' \
' 4. ping <src_vm_name> <dst_vm_name_or_ip> - 测试连通性\n' \
' 5. create_peering <name> <vpc1> <vpc2> - 创建对等连接\n' \
' 6. show - 查看 VPC 拓扑\n' \
' 7. exit - 退出并清理资源\n' \
'==========================================================='
prompt = '(VPC-Sim) '
def emptyline(self):
pass
def do_create_vpc(self, arg):
"""创建 VPC: create_vpc <name> <cidr>"""
args = shlex.split(arg)
if len(args) != 2: return print("❌ 用法: create_vpc <name> <cidr>")
try: VPC(args[0], args[1])
except Exception as e: print(f"❌ 创建失败: {e}")
def do_create_subnet(self, arg):
"""创建子网: create_subnet <vpc_name> <subnet_name> <cidr>"""
args = shlex.split(arg)
if len(args) != 3: return print("❌ 用法: create_subnet <vpc_name> <subnet_name> <cidr>")
vpc = ALL_VPCS.get(args[0])
if not vpc: return print(f"❌ VPC {args[0]} 不存在")
try: vpc.add_subnet(args[1], args[2])
except Exception as e: print(f"❌ 创建失败: {e}")
def do_create_vm(self, arg):
"""创建虚拟机: create_vm <subnet_cidr> <vm_name> [ip]"""
args = shlex.split(arg)
if len(args) < 2: return print("❌ 用法: create_vm <subnet_cidr> <vm_name> [ip]")
target_subnet = None
for vpc in ALL_VPCS.values():
for subnet in vpc.subnets:
if str(subnet.cidr) == args[0]:
target_subnet = subnet
break
if not target_subnet: return print(f"❌ 子网 {args[0]} 不存在")
ip = args[2] if len(args) > 2 else None
try: target_subnet.create_vm(args[1], ip)
except Exception as e: print(f"❌ 创建失败: {e}")
def do_ping(self, arg):
"""Ping 测试: ping <src_vm_name> <dst_vm_name_or_ip>"""
args = shlex.split(arg)
if len(args) != 2: return print("❌ 用法: ping <src_vm_name> <dst_vm_name_or_ip>")
target_vm = None
target_ip = args[1]
for vpc in ALL_VPCS.values():
for subnet in vpc.subnets:
for vm in subnet.vms:
if vm.name == args[0]: target_vm = vm
if vm.name == args[1]: target_ip = str(vm.ip)
if not target_vm: return print(f"❌ 源虚拟机 {args[0]} 不存在")
target_vm.ping(target_ip)
# ---------------- 对等连接管理 ----------------
def do_create_peering(self, arg):
"""创建 VPC 对等连接: create_peering <name> <vpc1> <vpc2>"""
args = shlex.split(arg)
if len(args) != 3: return print("❌ 用法: create_peering <name> <vpc1> <vpc2>")
v1, v2 = ALL_VPCS.get(args[1]), ALL_VPCS.get(args[2])
if not v1 or not v2: return print("❌ VPC 不存在")
ALL_PEERINGS[args[0]] = VPCPeering(args[0], v1, v2)
def do_show(self, arg):
"""查看 VPC 拓扑: show"""
print("\n=== 当前 VPC 拓扑 ===")
if not ALL_VPCS: print("空")
for vpc in ALL_VPCS.values():
print(f"🌐 VPC: {vpc.name} ({vpc.cidr})")
for subnet in vpc.subnets:
print(f" ├─ 📂 子网: {subnet.name} - {subnet.cidr} (网关: {subnet.gateway})")
for vm in subnet.vms:
print(f" │ └─ 💻 VM: {vm.name} (IP: {vm.ip})")
print("===================\n")
def do_exit(self, arg):
"""退出模拟器: exit"""
print("退出并清理资源...")
return True一键体验对等连接：
将以下指令粘贴进模拟器，可以直接复刻两个互通的 VPC 网络：
create_vpc prod 10.0.0.0/16
create_vpc dev 192.168.0.0/16
create_subnet prod prod-sub 10.0.1.0/24
create_subnet dev dev-sub 192.168.1.0/24
create_vm 10.0.1.0/24 prod-vm
create_vm 192.168.1.0/24 dev-vm
ping prod-vm dev-vm
create_peering peer-link prod dev
ping prod-vm dev-vmVPC 对等连接是两个 VPC 之间的私有网络连接，不经过公网，具有低延迟、高安全、免费的特点。它通过在两个 VPC 的路由表中添加对方的网段路由，实现跨 VPC 的流量转发。
四、完整可运行代码
import subprocess
import ipaddress
import cmd
import shlex
from typing import List, Dict, Optional
import atexit
# ------------------------------
# 全局资源管理
# ------------------------------
ALL_VPCS: Dict[str, 'VPC'] = {}
ALL_NETNS: List[str] = []
ALL_BRIDGES: List[str] = []
ALL_PEERINGS: Dict[str, 'VPCPeering'] = {}
def cleanup_all():
print("\n" + "="*50)
print("🔄 自动清理所有资源...")
# 清理网络命名空间和网桥
for netns in ALL_NETNS:
subprocess.run(f"ip netns del {netns} 2>/dev/null", shell=True, capture_output=True)
subprocess.run(f"ip link del veth-{netns} 2>/dev/null", shell=True, capture_output=True)
for bridge in ALL_BRIDGES:
subprocess.run(f"iptables -D FORWARD -i {bridge} -o {bridge} -j ACCEPT 2>/dev/null", shell=True, capture_output=True)
subprocess.run(f"iptables -D FORWARD -i {bridge} -o br-+ -j DROP 2>/dev/null", shell=True, capture_output=True)
subprocess.run(f"iptables -D FORWARD -o {bridge} -i br-+ -j DROP 2>/dev/null", shell=True, capture_output=True)
subprocess.run(f"ip link del {bridge} 2>/dev/null", shell=True, capture_output=True)
print("✅ 所有资源清理完成")
atexit.register(cleanup_all)
# ------------------------------
# VPC对等连接
# ------------------------------
class VPCPeering:
"""
模拟VPC对等连接：实现两个VPC之间的私有网络互通
对应云厂商控制台：VPC对等连接管理
"""
def __init__(self, name: str, vpc1: 'VPC', vpc2: 'VPC'):
self.name = name
self.vpc1 = vpc1
self.vpc2 = vpc2
# 添加互相的路由
vpc1.add_route(str(vpc2.cidr), vpc1.bridge_name)
vpc2.add_route(str(vpc1.cidr), vpc2.bridge_name)
# 允许两个VPC之间的流量转发
subprocess.run(
f"iptables -I FORWARD -s {vpc1.cidr} -d {vpc2.cidr} -j ACCEPT",
shell=True, check=True
)
subprocess.run(
f"iptables -I FORWARD -s {vpc2.cidr} -d {vpc1.cidr} -j ACCEPT",
shell=True, check=True
)
print(f"✅ 创建VPC对等连接 [{name}]：{vpc1.name} ↔ {vpc2.name}")
# ------------------------------
# 基础VPC类
# ------------------------------
class VPC:
def __init__(self, name: str, cidr: str):
self.name = name
self.cidr = ipaddress.IPv4Network(cidr)
self.subnets: List['Subnet'] = []
self.route_table: Dict[str, str] = {}
self.bridge_name = f"br-{self.name}"
self._create_bridge()
ALL_VPCS[name] = self
ALL_BRIDGES.append(self.bridge_name)
print(f"✅ 创建VPC [{self.name}]，网段：{self.cidr}")
def _create_bridge(self):
"""创建VPC内部的虚拟网桥（对应分布式虚拟交换机DVS）"""
subprocess.run(f"ip link del {self.bridge_name} 2>/dev/null", shell=True, capture_output=True)
subprocess.run(f"ip link add {self.bridge_name} type bridge", shell=True, check=True)
subprocess.run(f"ip link set {self.bridge_name} up", shell=True, check=True)
# 如果安装了 Docker，FORWARD 链默认会被设置为 DROP，导致跨子网不通
# 添加 iptables 规则：允许同 VPC 内部跨子网转发，同时隔离不同 VPC！
# 注意：使用 -I 插入时，后执行的命令会在链的最上方！所以先 DROP，后 ACCEPT
# 1. 隔离不同 VPC（丢弃跨网桥流量）
subprocess.run(f"iptables -I FORWARD -i {self.bridge_name} -o br-+ -j DROP", shell=True, check=True)
subprocess.run(f"iptables -I FORWARD -o {self.bridge_name} -i br-+ -j DROP", shell=True, check=True)
# 2. 允许同 VPC 内部互通（这条会在最顶部优先匹配）
subprocess.run(f"iptables -I FORWARD -i {self.bridge_name} -o {self.bridge_name} -j ACCEPT", shell=True, check=True)
def add_subnet(self, name: str, subnet_cidr: str) -> 'Subnet':
if not ipaddress.IPv4Network(subnet_cidr).subnet_of(self.cidr):
raise ValueError(f"子网 {subnet_cidr} 不属于VPC网段 {self.cidr}")
subnet = Subnet(name, subnet_cidr, self)
self.subnets.append(subnet)
self.route_table[subnet_cidr] = self.bridge_name
return subnet
def add_route(self, dest_cidr: str, next_hop: str):
"""添加自定义路由规则"""
self.route_table[dest_cidr] = next_hop
print(f"✅ 添加路由：{dest_cidr} -> {next_hop}")
class Subnet:
def __init__(self, name: str, cidr: str, vpc: VPC):
self.name = name
self.cidr = ipaddress.IPv4Network(cidr)
self.vpc = vpc
self.vms: List['VM'] = []
self.gateway = str(self.cidr[1])
self._configure_gateway()
print(f"✅ 创建子网 {self.name} [{self.cidr}]，网关：{self.gateway}")
def _configure_gateway(self):
subprocess.run(
f"ip addr del {self.gateway}/{self.cidr.prefixlen} dev {self.vpc.bridge_name} 2>/dev/null",
shell=True, capture_output=True
)
subprocess.run(
f"ip addr add {self.gateway}/{self.cidr.prefixlen} dev {self.vpc.bridge_name}",
shell=True, check=True
)
def create_vm(self, name: str, ip: Optional[str] = None) -> 'VM':
if not ip:
ip = str(self.cidr[len(self.vms)+2])
else:
if not ipaddress.IPv4Address(ip) in self.cidr:
raise ValueError(f"IP {ip} 不属于子网 {self.cidr}")
vm = VM(name, ip, self)
self.vms.append(vm)
return vm
class VM:
def __init__(self, name: str, ip: str, subnet: Subnet):
self.name = name
self.ip = ipaddress.IPv4Address(ip)
self.subnet = subnet
self.vpc = subnet.vpc
self.veth_name = f"veth-{self.name}"
self._create_netns()
self._connect_to_vpc()
ALL_NETNS.append(self.name)
print(f"✅ 创建虚拟机 [{self.name}]，IP：{self.ip}")
def _create_netns(self):
subprocess.run(f"ip netns del {self.name} 2>/dev/null", shell=True, capture_output=True)
subprocess.run(f"ip netns add {self.name}", shell=True, check=True)
subprocess.run(f"ip netns exec {self.name} ip link set lo up", shell=True, check=True)
def _connect_to_vpc(self):
subprocess.run(f"ip link del {self.veth_name} 2>/dev/null", shell=True, capture_output=True)
subprocess.run(
f"ip link add {self.veth_name} type veth peer name eth0 netns {self.name}",
shell=True, check=True
)
subprocess.run(
f"ip link set {self.veth_name} master {self.vpc.bridge_name}",
shell=True, check=True
)
subprocess.run(f"ip link set {self.veth_name} up", shell=True, check=True)
subprocess.run(
f"ip netns exec {self.name} ip addr add {self.ip}/{self.subnet.cidr.prefixlen} dev eth0",
shell=True, check=True
)
subprocess.run(f"ip netns exec {self.name} ip link set eth0 up", shell=True, check=True)
subprocess.run(
f"ip netns exec {self.name} ip route add default via {self.subnet.gateway}",
shell=True, check=True
)
def ping(self, target_ip: str, count: int = 2, timeout: int = 1) -> bool:
print(f"\n🔍 测试连通性：{self.name}({self.ip}) → {target_ip}")
result = subprocess.run(
f"ip netns exec {self.name} ping -c {count} -W {timeout} {target_ip}",
shell=True, capture_output=True, text=True
)
if result.returncode == 0:
print("✅ 连通成功！")
return True
else:
print("❌ 连通失败！")
return False
def exec(self, command: str) -> str:
result = subprocess.run(
f"ip netns exec {self.name} {command}",
shell=True, capture_output=True, text=True
)
return result.stdout
# ------------------------------
# 交互式 CLI Shell
# ------------------------------
class VPCShell(cmd.Cmd):
intro = '===========================================================\n' \
'欢迎使用 VPC 交互式模拟器（对等连接篇）\n' \
'【可用指令】:\n' \
' 1. create_vpc <name> <cidr> - 创建 VPC\n' \
' 2. create_subnet <vpc_name> <name> <cidr> - 创建子网\n' \
' 3. create_vm <subnet_cidr> <vm_name> [ip] - 创建虚拟机\n' \
' 4. ping <src_vm_name> <dst_vm_name_or_ip> - 测试连通性\n' \
' 5. create_peering <name> <vpc1> <vpc2> - 创建对等连接\n' \
' 6. show - 查看 VPC 拓扑\n' \
' 7. exit - 退出并清理资源\n' \
'==========================================================='
prompt = '(VPC-Sim) '
def emptyline(self):
pass
def do_create_vpc(self, arg):
"""创建 VPC: create_vpc <name> <cidr>"""
args = shlex.split(arg)
if len(args) != 2: return print("❌ 用法: create_vpc <name> <cidr>")
try: VPC(args[0], args[1])
except Exception as e: print(f"❌ 创建失败: {e}")
def do_create_subnet(self, arg):
"""创建子网: create_subnet <vpc_name> <subnet_name> <cidr>"""
args = shlex.split(arg)
if len(args) != 3: return print("❌ 用法: create_subnet <vpc_name> <subnet_name> <cidr>")
vpc = ALL_VPCS.get(args[0])
if not vpc: return print(f"❌ VPC {args[0]} 不存在")
try: vpc.add_subnet(args[1], args[2])
except Exception as e: print(f"❌ 创建失败: {e}")
def do_create_vm(self, arg):
"""创建虚拟机: create_vm <subnet_cidr> <vm_name> [ip]"""
args = shlex.split(arg)
if len(args) < 2: return print("❌ 用法: create_vm <subnet_cidr> <vm_name> [ip]")
target_subnet = None
for vpc in ALL_VPCS.values():
for subnet in vpc.subnets:
if str(subnet.cidr) == args[0]:
target_subnet = subnet
break
if not target_subnet: return print(f"❌ 子网 {args[0]} 不存在")
ip = args[2] if len(args) > 2 else None
try: target_subnet.create_vm(args[1], ip)
except Exception as e: print(f"❌ 创建失败: {e}")
def do_ping(self, arg):
"""Ping 测试: ping <src_vm_name> <dst_vm_name_or_ip>"""
args = shlex.split(arg)
if len(args) != 2: return print("❌ 用法: ping <src_vm_name> <dst_vm_name_or_ip>")
target_vm = None
target_ip = args[1]
for vpc in ALL_VPCS.values():
for subnet in vpc.subnets:
for vm in subnet.vms:
if vm.name == args[0]: target_vm = vm
if vm.name == args[1]: target_ip = str(vm.ip)
if not target_vm: return print(f"❌ 源虚拟机 {args[0]} 不存在")
target_vm.ping(target_ip)
def do_create_peering(self, arg):
"""创建 VPC 对等连接: create_peering <name> <vpc1> <vpc2>"""
args = shlex.split(arg)
if len(args) != 3: return print("❌ 用法: create_peering <name> <vpc1> <vpc2>")
v1, v2 = ALL_VPCS.get(args[1]), ALL_VPCS.get(args[2])
if not v1 or not v2: return print("❌ VPC 不存在")
ALL_PEERINGS[args[0]] = VPCPeering(args[0], v1, v2)
def do_show(self, arg):
"""查看 VPC 拓扑: show"""
print("\n=== 当前 VPC 拓扑 ===")
if not ALL_VPCS: print("空")
for vpc in ALL_VPCS.values():
print(f"🌐 VPC: {vpc.name} ({vpc.cidr})")
for subnet in vpc.subnets:
print(f" ├─ 📂 子网: {subnet.name} - {subnet.cidr} (网关: {subnet.gateway})")
for vm in subnet.vms:
print(f" │ └─ 💻 VM: {vm.name} (IP: {vm.ip})")
print("===================\n")
def do_exit(self, arg):
"""退出模拟器: exit"""
print("退出并清理资源...")
return True
if __name__ == "__main__":
# 确保宿主机开启内核 IP 转发（VPC 路由器的关键）
subprocess.run("sysctl -w net.ipv4.ip_forward=1", shell=True, capture_output=True)
VPCShell().cmdloop()五、运行方法
- 将代码保存为
vpc_part4_peering.py - 使用 sudo 权限执行：
sudo python3 vpc_part4_peering.py- 运行结果
root@ubuntu:~# python3 vpc_part4_peering.py
===========================================================
欢迎使用 VPC 交互式模拟器（对等连接篇）
【可用指令】:
1. create_vpc <name> <cidr> - 创建 VPC
2. create_subnet <vpc_name> <name> <cidr> - 创建子网
3. create_vm <subnet_cidr> <vm_name> [ip] - 创建虚拟机
4. ping <src_vm_name> <dst_vm_name_or_ip> - 测试连通性
5. create_peering <name> <vpc1> <vpc2> - 创建对等连接
6. show - 查看 VPC 拓扑
7. exit - 退出并清理资源
===========================================================
(VPC-Sim) create_vpc prod 10.0.0.0/16
✅ 创建VPC [prod]，网段：10.0.0.0/16
(VPC-Sim) create_vpc dev 192.168.0.0/16
✅ 创建VPC [dev]，网段：192.168.0.0/16
(VPC-Sim) create_subnet prod prod-sub 10.0.1.0/24
✅ 创建子网 prod-sub [10.0.1.0/24]，网关：10.0.1.1
(VPC-Sim) create_subnet dev dev-sub 192.168.1.0/24
✅ 创建子网 dev-sub [192.168.1.0/24]，网关：192.168.1.1
(VPC-Sim) create_vm 10.0.1.0/24 prod-vm
✅ 创建虚拟机 [prod-vm]，IP：10.0.1.2
(VPC-Sim) create_vm 192.168.1.0/24 dev-vm
✅ 创建虚拟机 [dev-vm]，IP：192.168.1.2
(VPC-Sim) ping prod-vm dev-vm
🔍 测试连通性：prod-vm(10.0.1.2) → 192.168.1.2
❌ 连通失败！
(VPC-Sim) create_peering peer-link prod dev
✅ 添加路由：192.168.0.0/16 -> br-prod
✅ 添加路由：10.0.0.0/16 -> br-dev
✅ 创建VPC对等连接 [peer-link]：prod ↔ dev
(VPC-Sim) ping prod-vm dev-vm
🔍 测试连通性：prod-vm(10.0.1.2) → 192.168.1.2
✅ 连通成功！六、VPC 对等连接架构图
七、在线互动体验
用下面的交互式模拟器，亲手体验对等连接的建立过程和连通性变化：
🎮 动手试试：VPC 对等连接
先测试隔离状态，再建立对等连接，观察连通性变化
常见问题
两个 VPC 的网段可以重叠吗？
绝对不可以。如果两个 VPC 的网段重叠，路由表会发生冲突，对等连接将无法正常工作。这是云平台 VPC 设计的基本原则。
对等连接是单向的还是双向的？
是双向的。一旦创建成功，两个 VPC 之间可以自由通信，不需要额外配置单向规则。
真实云平台中对等连接可以跨账号吗？
可以。大多数云厂商支持跨账号的 VPC 对等连接，但需要双方账号都同意接受连接请求。我们的模拟器简化了账号概念，只需要传入两个 VPC 对象即可。
对等连接可以传递吗？
不可以。如果 VPC A 和 VPC B 建立了对等连接，VPC B 和 VPC C 建立了对等连接，那么 VPC A 和 VPC C 之间不能直接通信，需要单独建立 A 和 C 之间的对等连接。
下一篇预告
最后一篇我们将深入云厂商真实 VPC 的实现原理，从单机模拟器过渡到大规模分布式 SDN 架构，讲解 VXLAN 隧道技术和云网络的性能优化。