Python运维开发面试题

运维开发（DevOps/开发运维）岗位除了Python基础外，更注重系统管理、自动化、监控、容器化等方面的知识。

一、系统管理与Shell交互

1. 执行系统命令

• 题目：Python中调用Shell命令有哪几种方式？区别是什么？
• 考察点：系统交互能力。
• 答案：

  # os.system - 简单，返回退出码
  os.system("ls -l")
  
  # os.popen - 可以读取输出（旧式）
  result = os.popen("ls -l").read()
  
  # subprocess - 推荐，功能强大
  import subprocess
  # 执行命令获取输出
  result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
  print(result.stdout)
  
  # 实时输出
  proc = subprocess.Popen(["ping", "baidu.com"], stdout=subprocess.PIPE)
  for line in proc.stdout:
      print(line.decode().strip())
  

2. 获取系统信息

• 题目：如何获取当前主机的CPU、内存、磁盘使用情况？
• 考察点：系统监控基础。
• 答案：

  import psutil  # 第三方库，运维开发必备
  
  # CPU
  cpu_percent = psutil.cpu_percent(interval=1)
  
  # 内存
  mem = psutil.virtual_memory()
  mem_percent = mem.percent
  
  # 磁盘
  disk = psutil.disk_usage('/')
  disk_percent = disk.used / disk.total * 100
  
  # 不使用psutil（Linux系统）
  with open('/proc/meminfo') as f:
      mem_info = f.readlines()
  

3. 环境变量

• 题目：如何在Python中读取和设置环境变量？
• 考察点：配置管理。
• 答案：

  import os
  
  # 读取
  db_host = os.environ.get('DB_HOST', 'localhost')
  
  # 设置（仅影响当前进程）
  os.environ['MY_VAR'] = 'value'
  

二、文件与目录操作

4. 遍历目录

• 题目：如何递归查找某个目录下所有 .log 文件？
• 考察点：文件系统操作。
• 答案：

  import os
  
  # 方法1：os.walk
  log_files = []
  for root, dirs, files in os.walk('/var/log'):
      for file in files:
          if file.endswith('.log'):
              log_files.append(os.path.join(root, file))
  
  # 方法2：pathlib (Python 3.4+)
  from pathlib import Path
  log_files = list(Path('/var/log').rglob('*.log'))
  

5. 文件监控

• 题目：如何监控一个文件是否被修改？
• 考察点：实时处理能力。
• 答案：

  # 简单方法：记录文件修改时间
  import os
  import time
  
  def watch_file(filename):
      last_mtime = os.path.getmtime(filename)
      while True:
          time.sleep(1)
          current_mtime = os.path.getmtime(filename)
          if current_mtime != last_mtime:
              print("文件已修改")
              last_mtime = current_mtime
  
  # 专业方法：使用 watchdog 库
  from watchdog.observers import Observer
  from watchdog.events import FileSystemEventHandler
  

6. 大文件处理

• 题目：如何处理几个GB的日志文件，避免内存溢出？
• 考察点：内存优化。
• 答案：

  # 逐行读取，不要用read()
  with open('large.log', 'r') as f:
      for line in f:
          process(line)  # 逐行处理
  
  # 使用生成器
  def read_large_file(file_path):
      with open(file_path, 'r') as f:
          for line in f:
              yield line.strip()
  

三、网络与HTTP

7. HTTP请求

• 题目：使用Python发送HTTP GET/POST请求，如何处理超时和重试？
• 考察点：API调用能力。
• 答案：

  import requests
  from requests.adapters import HTTPAdapter
  from urllib3.util.retry import Retry
  
  # 基本用法
  response = requests.get('http://api.example.com', timeout=5)
  response = requests.post('http://api.example.com', json={'key': 'value'})
  
  # 带重试的会话
  session = requests.Session()
  retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503])
  adapter = HTTPAdapter(max_retries=retry)
  session.mount('http://', adapter)
  session.mount('https://', adapter)
  

8. 端口检查

• 题目：如何检查远程主机的某个端口是否开放？
• 考察点：网络连通性检测。
• 答案：

  import socket
  
  def check_port(host, port, timeout=3):
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      sock.settimeout(timeout)
      result = sock.connect_ex((host, port))
      sock.close()
      return result == 0  # 0表示端口开放
  

9. 获取公网IP

• 题目：如何获取当前机器的公网IP地址？
• 考察点：实际应用场景。
• 答案：

  import requests
  
  # 使用第三方API
  def get_public_ip():
      try:
          ip = requests.get('https://api.ipify.org', timeout=5).text
          return ip
      except:
          return None
  

四、并发与性能

10. 批量任务处理

• 题目：需要对1000台服务器执行命令，如何提高效率？
• 考察点：并发设计。
• 答案：

  import concurrent.futures
  import paramiko  # SSH库
  
  def exec_on_host(host, command):
      ssh = paramiko.SSHClient()
      ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
      ssh.connect(host, username='root')
      stdin, stdout, stderr = ssh.exec_command(command)
      return host, stdout.read().decode()
  
  hosts = ['server{}'.format(i) for i in range(1000)]
  
  # 使用线程池并发执行
  with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
      future_to_host = {executor.submit(exec_on_host, host, 'uptime'): host 
                        for host in hosts}
      for future in concurrent.futures.as_completed(future_to_host):
          host, result = future.result()
          print(f"{host}: {result}")
  

11. 异步IO

• 题目：解释异步IO在运维开发中的应用场景。
• 考察点：对异步的理解。
• 答案：

  # 适合大量网络请求的场景，如健康检查
  import asyncio
  import aiohttp
  
  async def check_health(url):
      async with aiohttp.ClientSession() as session:
          try:
              async with session.get(url, timeout=5) as resp:
                  return url, resp.status
          except:
              return url, 'DOWN'
  
  async def main():
      urls = ['http://service1', 'http://service2', ...]
      tasks = [check_health(url) for url in urls]
      results = await asyncio.gather(*tasks)
      for url, status in results:
          print(f"{url}: {status}")
  

五、容器与编排

12. Docker操作

• 题目：如何在Python中管理Docker容器？
• 考察点：容器化技能。
• 答案：

  # 使用docker-py库
  import docker
  
  client = docker.from_env()
  
  # 运行容器
  container = client.containers.run('nginx:latest', 
                                    detach=True, 
                                    ports={'80/tcp': 8080})
  
  # 获取容器列表
  containers = client.containers.list()
  
  # 执行命令
  result = container.exec_run('nginx -t')
  
  # 清理
  container.stop()
  container.remove()
  

13. Kubernetes操作

• 题目：如何通过Python获取K8s集群的Pod状态？
• 考察点：容器编排能力。
• 答案：

  from kubernetes import client, config
  
  # 加载kube配置
  config.load_kube_config()
  
  # 创建API客户端
  v1 = client.CoreV1Api()
  
  # 获取所有Pod
  pods = v1.list_pod_for_all_namespaces(watch=False)
  for pod in pods.items:
      print(f"{pod.metadata.namespace}\t{pod.metadata.name}\t{pod.status.phase}")
  
  # 获取特定命名空间的Pod
  namespace_pods = v1.list_namespaced_pod(namespace='default')
  

六、日志处理与分析

14. 日志解析

• 题目：如何解析Nginx日志并统计IP访问次数？
• 考察点：日志分析能力。
• 答案：

  import re
  from collections import Counter
  
  log_pattern = r'(\d+\.\d+\.\d+\.\d+).*\[(.*?)\].*"(.*?)".*(\d+)'
  ip_counter = Counter()
  
  with open('access.log', 'r') as f:
      for line in f:
          match = re.search(log_pattern, line)
          if match:
              ip = match.group(1)
              ip_counter[ip] += 1
  
  # 输出TOP10
  for ip, count in ip_counter.most_common(10):
      print(f"{ip}: {count}")
  

15. 日志轮转

• 题目：如何实现Python日志的自动切割？
• 考察点：日志管理。
• 答案：

  import logging
  from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
  
  # 按大小切割（10MB，保留5个备份）
  handler = RotatingFileHandler('app.log', maxBytes=10*1024*1024, backupCount=5)
  
  # 按时间切割（每天，保留30天）
  handler = TimedRotatingFileHandler('app.log', when='midnight', 
                                     interval=1, backupCount=30)
  
  logging.basicConfig(handlers=[handler], level=logging.INFO)
  

七、自动化运维工具

16. Ansible/Fabric

• 题目：使用Fabric库实现远程执行命令和文件传输。
• 考察点：自动化工具。
• 答案：

  # Fabric 2.x
  from fabric import Connection
  
  def deploy():
      c = Connection('web-server')
      # 执行命令
      result = c.run('uname -s', hide=True)
      print(result.stdout)
      
      # 上传文件
      c.put('local_file.txt', '/remote/file.txt')
      
      # 下载文件
      c.get('/remote/log.log', 'local_log.log')
  

17. 配置管理

• 题目：如何处理不同环境的配置（开发、测试、生产）？
• 考察点：配置管理经验。
• 答案：

  import os
  from pathlib import Path
  
  class Config:
      """配置基类"""
      DEBUG = False
      TESTING = False
      DATABASE_URI = 'sqlite:///:memory:'
  
  class DevelopmentConfig(Config):
      DEBUG = True
      DATABASE_URI = 'sqlite:///dev.db'
  
  class ProductionConfig(Config):
      DATABASE_URI = os.environ.get('DATABASE_URL', 'postgresql://...')
  
  # 根据环境变量加载配置
  config_map = {
      'development': DevelopmentConfig,
      'production': ProductionConfig
  }
  
  env = os.environ.get('FLASK_ENV', 'development')
  config = config_map[env]()
  

八、监控与告警

18. 自定义监控脚本

• 题目：写一个监控脚本，检查Web服务是否正常，异常时发送告警。
• 考察点：监控系统设计。
• 答案：

  import requests
  import smtplib
  import time
  from email.mime.text import MIMEText
  
  def check_service(url):
      try:
          resp = requests.get(url, timeout=5)
          return resp.status_code == 200
      except:
          return False
  
  def send_alert(service_name):
      msg = MIMEText(f"{service_name} 服务异常！")
      msg['Subject'] = '服务告警'
      msg['From'] = 'monitor@example.com'
      msg['To'] = 'admin@example.com'
      
      # 发送邮件（简化版）
      # ...
  
  def monitor():
      services = {
          'web': 'http://localhost:8080/health',
          'api': 'http://localhost:5000/health'
      }
      
      while True:
          for name, url in services.items():
              if not check_service(url):
                  send_alert(name)
          time.sleep(60)  # 每分钟检查一次
  

19. 指标收集

• 题目：如何收集系统指标并暴露给Prometheus？
• 考察点：监控系统集成。
• 答案：

  from prometheus_client import start_http_server, Gauge, Counter
  import psutil
  import time
  
  # 定义指标
  cpu_usage = Gauge('cpu_usage_percent', 'CPU使用率')
  memory_usage = Gauge('memory_usage_percent', '内存使用率')
  request_count = Counter('http_requests_total', 'HTTP请求总数')
  
  def collect_metrics():
      while True:
          cpu_usage.set(psutil.cpu_percent())
          memory_usage.set(psutil.virtual_memory().percent)
          time.sleep(15)
  
  if __name__ == '__main__':
      # 启动HTTP服务，暴露指标
      start_http_server(8000)
      collect_metrics()
  

九、数据库与缓存

20. Redis操作

• 题目：使用Redis实现一个简单的任务队列。
• 考察点：缓存/队列应用。
• 答案：

  import redis
  
  r = redis.Redis(host='localhost', port=6379, db=0)
  
  # 生产者
  def add_task(task_data):
      r.rpush('task_queue', task_data)
  
  # 消费者
  def process_task():
      while True:
          # 阻塞等待任务
          task = r.blpop('task_queue', timeout=0)
          if task:
              data = task[1].decode()
              print(f"处理任务: {data}")
              # 处理任务...
  

十、面试常见场景题

21. 需求：日志收集系统

• 题目：如果要设计一个简单的日志收集系统，从多台服务器收集日志到中心服务器，你会怎么设计？
• 考察点：系统设计能力。
• 答案要点：
  • 客户端：使用 rsyslog 或自定义Agent发送
  • 传输：使用TCP/UDP，考虑可靠性和压缩
  • 服务端：接收并写入消息队列（Kafka/RabbitMQ）
  • 处理：Logstash/Fluentd 解析
  • 存储：Elasticsearch
  • 展示：Kibana

22. 需求：自动化部署

• 题目：描述一下你设计过的自动化部署流程。
• 考察点：CI/CD经验。
• 答案要点：
  • 代码托管：Git
  • 触发：Webhook
  • 构建：Jenkins/GitLab CI
  • 测试：单元测试、集成测试
  • 打包：Docker镜像
  • 部署：Kubernetes滚动更新
  • 回滚：保留历史版本

23. 故障排查

• 题目：线上服务CPU飙升到100%，如何排查？
• 考察点：问题定位能力。
• 答案：

  # 1. top 找出高CPU进程PID
  top
  
  # 2. 查看进程线程
  top -H -p <PID>
  
  # 3. 转换为16进制
  printf "%x\n" <thread_id>
  
  # 4. 获取堆栈信息
  jstack <PID> | grep -A 10 <hex_thread_id>
  
  # 或使用Python的traceback
  import traceback
  import sys
  def print_stack():
      for thread_id, stack in sys._current_frames().items():
          print(f"Thread {thread_id}:")
          traceback.print_stack(stack)
  

十一、常用工具库

运维开发需要熟悉的Python库：

类别	库名	用途
系统	psutil	系统信息监控
网络	requests	HTTP请求
SSH	paramiko, fabric	远程执行命令
容器	docker-py	Docker管理
K8s	kubernetes	K8s API操作
数据库	pymysql, redis-py	数据库连接
并发	concurrent.futures	线程/进程池
配置	PyYAML	YAML解析
监控	prometheus_client	指标暴露

---

总结：运维开发岗位的面试更注重实际问题解决能力，建议准备时多思考：

1. 你遇到过什么线上问题？怎么解决的？
2. 如何实现某个自动化需求？
3. 系统架构怎么设计？

如果还需要某个方向的更深入题目，随时告诉我！