Cloud DevOps 面试准备¶

# 1. 两数之和
def two_sum(nums, target):
    """两数之和"""
    seen = {}
    for i, num in enumerate(nums):
        if target - num in seen:
            return [seen[target - num], i]
        seen[num] = i
    return []

# 2. 最长无重复子串
def length_of_longest_substring(s):
    """最长无重复子串"""
    char_map = {}
    left = 0
    max_len = 0

    for right, char in enumerate(s):
        if char in char_map and char_map[char] >= left:
            left = char_map[char] + 1
        char_map[char] = right
        max_len = max(max_len, right - left + 1)

    return max_len

# 3. 字符串反转
def reverse_string(s):
    """字符串反转"""
    return s[::-1]

# 1. 两数之和
def two_sum(nums, target):
    """两数之和"""
    seen = {}
    for i, num in enumerate(nums):
        if target - num in seen:
            return [seen[target - num], i]
        seen[num] = i
    return []

# 2. 移除元素
def remove_element(nums, val):
    """移除元素"""
    i = 0
    for num in nums:
        if num != val:
            nums[i] = num
            i += 1
    return i

# 3. 合并有序数组
def merge_sorted_arrays(nums1, nums2):
    """合并有序数组"""
    result = []
    i = j = 0

    while i < len(nums1) and j < len(nums2):
        if nums1[i] < nums2[j]:
            result.append(nums1[i])
            i += 1
        else:
            result.append(nums2[j])
            j += 1

    result.extend(nums1[i:])
    result.extend(nums2[j:])

    return result

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next

def reverse_linked_list(head):
    """反转链表"""
    prev = None
    curr = head

    while curr:
        next_temp = curr.next
        curr.next = prev
        prev = curr
        curr = next_temp

    return prev

def has_cycle(head):
    """检测链表环"""
    slow = fast = head

    while fast and fast.next:
        slow = slow.next
        fast = fast.next.next

        if slow == fast:
            return True

    return False

class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

def inorder_traversal(root):
    """中序遍历"""
    result = []

    def traverse(node):
        if not node:
            return
        traverse(node.left)
        result.append(node.val)
        traverse(node.right)

    traverse(root)
    return result

def max_depth(root):
    """最大深度"""
    if not root:
        return 0

    left_depth = max_depth(root.left)
    right_depth = max_depth(root.right)

    return max(left_depth, right_depth) + 1

class URLShortener:
    """短链接服务"""

    def __init__(self):
        self.url_map = {}
        self.counter = 0
        self.base = 62
        self.chars = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"

    def encode(self, num):
        """数字转短码"""
        if num == 0:
            return self.chars[0]

        result = []
        while num > 0:
            num, remainder = divmod(num, self.base)
            result.append(self.chars[remainder])

        return ''.join(reversed(result))

    def decode(self, short_code):
        """短码转数字"""
        num = 0
        for char in short_code:
            num = num * self.base + self.chars.index(char)
        return num

    def shorten(self, long_url):
        """生成短链接"""
        if long_url in self.url_map:
            return self.url_map[long_url]['short']

        self.counter += 1
        short_code = self.encode(self.counter)
        short_url = f"https://short.ly/{short_code}"

        self.url_map[long_url] = {
            'short': short_url,
            'code': short_code,
            'created_at': datetime.now()
        }

        return short_url

    def expand(self, short_url):
        """展开短链接"""
        short_code = short_url.split('/')[-1]
        num = self.decode(short_code)

        for long_url, data in self.url_map.items():
            if data['code'] == short_code:
                return long_url

        return None

import time
from collections import deque

class RateLimiter:
    """限流器"""

    def __init__(self, rate, capacity):
        """
        rate: 速率（请求/秒）
        capacity: 桶容量
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
        self.requests = deque()

    def allow(self):
        """检查是否允许请求"""
        now = time.time()

        # 计算时间差
        time_passed = now - self.last_update
        self.last_update = now

        # 添加令牌
        self.tokens = min(self.capacity, self.tokens + time_passed * self.rate)

        # 清理过期请求
        while self.requests and self.requests[0] <= now - 1:
            self.requests.popleft()

        # 检查是否允许
        if self.tokens >= 1 and len(self.requests) < self.capacity:
            self.tokens -= 1
            self.requests.append(now)
            return True

        return False

from flask import Flask, jsonify
import redis
import json
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)

# Redis缓存
redis_client = redis.Redis(host='redis', port=6379, db=0)

# 线程池
executor = ThreadPoolExecutor(max_workers=10)

@app.route('/api/data/<data_id>')
def get_data(data_id):
    """获取数据"""
    # 尝试从缓存获取
    cache_key = f"data:{data_id}"
    cached = redis_client.get(cache_key)

    if cached:
        return jsonify({'source': 'cache', 'data': json.loads(cached)})

    # 从数据库获取
    data = get_from_database(data_id)

    # 异步更新缓存
    executor.submit(update_cache, cache_key, data)

    return jsonify({'source': 'database', 'data': data})

def update_cache(cache_key, data):
    """更新缓存"""
    redis_client.setex(cache_key, 300, json.dumps(data))

def get_from_database(data_id):
    """从数据库获取"""
    # 实际数据库查询
    return {'id': data_id, 'value': 'data'}