互联网大厂面试题100道-阿里百度篇-完整版
·
互联网大厂经典面试题100道 - 阿里、百度篇
📋 总体提纲
🎯 第一部分：编程基础 (30题)
- 数据结构与算法 (15题)：树、图、动态规划、字符串处理
- 编程语言 (10题)：Java/Python/C++、JVM、并发编程
- 设计模式 (5题)：工厂、代理、策略、模板方法
🔧 第二部分：系统设计 (25题)
- 分布式系统 (10题)：CAP理论、一致性算法、分布式锁
- 数据库设计 (8题)：分库分表、读写分离、缓存策略
- 搜索引擎 (7题)：倒排索引、相关性排序、爬虫架构
🌐 第三部分：公司特色题目 (45题)
- 阿里特色 (25题)：电商架构、云计算、中台战略、双11
- 百度特色 (20题)：搜索引擎、AI应用、地图服务、自动驾驶
📝 详细题目与答案
🎯 第一部分：编程基础 (30题)
数据结构与算法 (15题)
1. 实现一个高效的布隆过滤器
import mmh3
import bitarray
from typing import List, Any
class BloomFilter:
def __init__(self, capacity: int, error_rate: float = 0.001):
"""
初始化布隆过滤器
capacity: 预期元素数量
error_rate: 期望错误率
"""
self.capacity = capacity
self.error_rate = error_rate
# 计算最优的bit数组大小和hash函数数量
self.bit_size = self._calculate_bit_size(capacity, error_rate)
self.hash_count = self._calculate_hash_count(self.bit_size, capacity)
# 初始化bit数组
self.bit_array = bitarray.bitarray(self.bit_size)
self.bit_array.setall(0)
# 元素计数
self.count = 0
def _calculate_bit_size(self, capacity: int, error_rate: float) -> int:
"""计算bit数组大小"""
import math
bit_size = -capacity * math.log(error_rate) / (math.log(2) ** 2)
return int(bit_size)
def _calculate_hash_count(self, bit_size: int, capacity: int) -> int:
"""计算hash函数数量"""
import math
hash_count = (bit_size / capacity) * math.log(2)
return max(1, int(hash_count))
def _get_hash_values(self, item: Any) -> List[int]:
"""获取多个hash值"""
hash_values = []
for i in range(self.hash_count):
# 使用不同的种子生成hash值
hash_val = mmh3.hash(str(item), i) % self.bit_size
hash_values.append(abs(hash_val))
return hash_values
def add(self, item: Any) -> None:
"""添加元素"""
hash_values = self._get_hash_values(item)
for hash_val in hash_values:
self.bit_array[hash_val] = 1
self.count += 1
def __contains__(self, item: Any) -> bool:
"""检查元素是否存在"""
hash_values = self._get_hash_values(item)
return all(self.bit_array[hash_val] for hash_val in hash_values)
def __len__(self) -> int:
"""返回已添加元素数量"""
return self.count
def false_positive_rate(self) -> float:
"""计算当前错误率"""
import math
if self.count == 0:
return 0.0
# (1 - e^(-kn/m))^k
rate = (1 - math.exp(-self.hash_count * self.count / self.bit_size)) ** self.hash_count
return rate
答案解析：
- 布隆过滤器使用多个哈希函数和位数组实现高效的存在性判断
- 时间复杂度O(k)，空间复杂度O(m)，其中k是哈希函数数量，m是位数组大小
- 支持分布式场景，使用Redis存储位数组
- 存在误判但不会漏判，适合缓存穿透、去重等场景
2. 实现一致性哈希算法
import hashlib
from typing import List, Dict, Tuple, Optional
import bisect
class ConsistentHash:
def __init__(self, nodes: List[str] = None, replicas: int = 150):
"""
初始化一致性哈希
nodes: 节点列表
replicas: 每个节点的虚拟节点数量
"""
self.replicas = replicas
self.ring = {} # 哈希环
self.sorted_keys = [] # 排序后的哈希值
self.nodes = set()
if nodes:
for node in nodes:
self.add_node(node)
def _hash(self, key: str) -> int:
"""计算hash值"""
return int(hashlib.md5(key.encode()).hexdigest(), 16)
def add_node(self, node: str) -> None:
"""添加节点"""
if node in self.nodes:
return
self.nodes.add(node)
# 为每个节点创建多个虚拟节点
for i in range(self.replicas):
virtual_key = f"{node}:{i}"
hash_value = self._hash(virtual_key)
self.ring[hash_value] = node
# 重新排序
self.sorted_keys = sorted(self.ring.keys())
def remove_node(self, node: str) -> None:
"""移除节点"""
if node not in self.nodes:
return
self.nodes.remove(node)
# 移除所有虚拟节点
keys_to_remove = []
for hash_value, node_name in self.ring.items():
if node_name == node:
keys_to_remove.append(hash_value)
for key in keys_to_remove:
del self.ring[key]
# 重新排序
self.sorted_keys = sorted(self.ring.keys())
def get_node(self, key: str) -> Optional[str]:
"""获取key对应的节点"""
if not self.ring:
return None
hash_value = self._hash(key)
# 找到第一个大于等于hash值的节点
index = bisect.bisect_right(self.sorted_keys, hash_value)
# 如果没找到，返回第一个节点（环形结构）
if index == len(self.sorted_keys):
index = 0
return self.ring[self.sorted_keys[index]]
def get_nodes(self, key: str, count: int) -> List[str]:
"""获取key对应的多个节点（用于副本）"""
if not self.ring or count <= 0:
return []
hash_value = self._hash(key)
index = bisect.bisect_right(self.sorted_keys, hash_value)
nodes = []
seen_nodes = set()
# 从当前节点开始，顺时针查找
for i in range(count):
current_index = (index + i) % len(self.sorted_keys)
node = self.ring[self.sorted_keys[current_index]]
if node not in seen_nodes:
nodes.append(node)
seen_nodes.add(node)
# 如果已经遍历完所有节点还没找到足够的节点
if len(seen_nodes) == len(self.nodes):
break
return nodes
答案解析：
- 一致性哈希解决分布式系统中数据分布和节点动态变化问题
- 通过虚拟节点提高数据分布的均匀性
- 节点增删时只影响相邻节点，最小化数据迁移
- 支持副本机制，提高系统可用性
3. 实现Trie树（前缀树）
from typing import Dict, List, Optional, Set
class TrieNode:
def __init__(self):
self.children: Dict[str, 'TrieNode'] = {}
self.is_end_of_word: bool = False
self.frequency: int = 0 # 词频统计
self.value: any = None # 存储额外值
class Trie:
def __init__(self):
self.root = TrieNode()
self.size = 0
def insert(self, word: str, value: any = None) -> None:
"""插入单词"""
node = self.root
for char in word:
if char not in node.children:
node.children[char] = TrieNode()
node = node.children[char]
if not node.is_end_of_word:
self.size += 1
node.is_end_of_word = True
node.frequency += 1
node.value = value
def search(self, word: str) -> bool:
"""搜索单词是否存在"""
node = self.root
for char in word:
if char not in node.children:
return False
node = node.children[char]
return node.is_end_of_word
def starts_with(self, prefix: str) -> bool:
"""检查是否有单词以该前缀开头"""
node = self.root
for char in prefix:
if char not in node.children:
return False
node = node.children[char]
return True
def get_all_words_with_prefix(self, prefix: str) -> List[str]:
"""获取所有以该前缀开头的单词"""
node = self.root
for char in prefix:
if char not in node.children:
return []
node = node.children[char]
words = []
self._dfs_collect_words(node, prefix, words)
return words
def _dfs_collect_words(self, node: TrieNode, current_word: str, words: List[str]) -> None:
"""DFS收集所有单词"""
if node.is_end_of_word:
words.append(current_word)
for char, child_node in node.children.items():
self._dfs_collect_words(child_node, current_word + char, words)
答案解析：
- Trie树专门用于字符串前缀匹配，时间复杂度O(m)，m为字符串长度
- 支持高效的前缀搜索、词频统计、自动补全等功能
- 空间复杂度较高，但可以通过压缩Trie优化
- 适用于搜索引擎、输入法、字典等场景
4. 实现跳表（Skip List）
import random
from typing import Optional, List, Any
class SkipListNode:
def __init__(self, value: Any, level: int):
self.value = value
self.forward = [None] * (level + 1)
class SkipList:
def __init__(self, max_level: int = 16, p: float = 0.5):
"""
初始化跳表
max_level: 最大层数
p: 节点提升到上一层的概率
"""
self.max_level = max_level
self.p = p
self.level = 0 # 当前最大层数
self.header = SkipListNode(float('-inf'), max_level)
def _random_level(self) -> int:
"""随机生成层数"""
level = 0
while random.random() < self.p and level < self.max_level:
level += 1
return level
def insert(self, value: Any) -> None:
"""插入元素"""
update = [None] * (self.max_level + 1)
current = self.header
# 从最高层开始查找插入位置
for i in range(self.level, -1, -1):
while current.forward[i] and current.forward[i].value < value:
current = current.forward[i]
update[i] = current
current = current.forward[0]
# 如果元素已存在，不插入
if current and current.value == value:
return
# 随机生成新节点的层数
new_level = self._random_level()
# 如果新节点的层数大于当前最大层数，更新update数组
if new_level > self.level:
for i in range(self.level + 1, new_level + 1):
update[i] = self.header
self.level = new_level
# 创建新节点
new_node = SkipListNode(value, new_level)
# 更新前驱节点的forward指针
for i in range(new_level + 1):
new_node.forward[i] = update[i].forward[i]
update[i].forward[i] = new_node
def search(self, value: Any) -> bool:
"""搜索元素"""
current = self.header
# 从最高层开始搜索
for i in range(self.level, -1, -1):
while current.forward[i] and current.forward[i].value < value:
current = current.forward[i]
current = current.forward[0]
return current is not None and current.value == value
答案解析：
- 跳表是一种概率性数据结构，支持快速搜索、插入、删除
- 平均时间复杂度O(log n)，空间复杂度O(n)
- 相比红黑树实现更简单，支持并发操作
- 广泛应用于Redis、LevelDB等数据库中
5. 实现LRU-K缓存算法
from collections import defaultdict, deque
import time
from typing import Any, Dict, List, Optional
class LRUKNode:
def __init__(self, key: Any, value: Any, k: int):
self.key = key
self.value = value
self.k = k
self.access_history = deque(maxlen=k) # 存储最近k次访问时间
self.last_access_time = 0
self.is_in_buffer = False
class LRUKCache:
def __init__(self, capacity: int, k: int = 2):
"""
LRU-K缓存
capacity: 缓存容量
k: 考虑的最近访问次数
"""
self.capacity = capacity
self.k = k
self.cache: Dict[Any, LRUKNode] = {}
self.history_order = deque() # 历史访问顺序
self.buffer_order = deque() # 缓存中的访问顺序
self.current_time = 0
def get(self, key: Any) -> Any:
"""获取缓存值"""
current_time = self._get_current_time()
if key not in self.cache:
return None
node = self.cache[key]
node.access_history.append(current_time)
node.last_access_time = current_time
# 如果访问次数达到k次且不在缓存中，移入缓存
if len(node.access_history) >= self.k and not node.is_in_buffer:
self._promote_to_buffer(node)
elif node.is_in_buffer:
# 更新缓存中的访问顺序
self._update_buffer_order(node)
return node.value
def put(self, key: Any, value: Any) -> None:
"""设置缓存值"""
current_time = self._get_current_time()
if key in self.cache:
# 更新现有节点
node = self.cache[key]
node.value = value
node.access_history.append(current_time)
node.last_access_time = current_time
if len(node.access_history) >= self.k and not node.is_in_buffer:
self._promote_to_buffer(node)
elif node.is_in_buffer:
self._update_buffer_order(node)
else:
# 创建新节点
node = LRUKNode(key, value, self.k)
node.access_history.append(current_time)
node.last_access_time = current_time
self.cache[key] = node
# 如果访问次数达到k次，尝试移入缓存
if len(node.access_history) >= self.k:
self._promote_to_buffer(node)
else:
# 添加到历史记录
self.history_order.append(key)
答案解析：
- LRU-K考虑最近K次访问时间，比传统LRU更准确
- 通过历史记录和缓存区分冷热数据
- 避免缓存污染，提高命中率
- 适用于数据库缓冲池、操作系统页面置换等场景
6. 实现Raft一致性算法
import time
import random
import threading
from enum import Enum
from typing import Dict, List, Optional, Any
class NodeState(Enum):
FOLLOWER = "follower"
CANDIDATE = "candidate"
LEADER = "leader"
class LogEntry:
def __init__(self, term: int, index: int, command: Any):
self.term = term
self.index = index
self.command = command
class RaftNode:
def __init__(self, node_id: str, peers: List[str]):
self.node_id = node_id
self.peers = peers
# 状态
self.state = NodeState.FOLLOWER
self.current_term = 0
self.voted_for: Optional[str] = None
self.log: List[LogEntry] = []
self.commit_index = 0
self.last_applied = 0
# Leader状态
self.next_index: Dict[str, int] = {}
self.match_index: Dict[str, int] = {}
# 定时器
self.election_timeout = random.uniform(5.0, 10.0)
self.heartbeat_interval = 2.0
self.last_heartbeat = time.time()
# 线程锁
self.lock = threading.RLock()
# 状态机
self.state_machine = {}
def start_election(self) -> None:
"""开始选举"""
with self.lock:
self.state = NodeState.CANDIDATE
self.current_term += 1
self.voted_for = self.node_id
print(f"Node {self.node_id} 开始选举，term {self.current_term}")
# 给自己投票
votes = 1
total_nodes = len(self.peers) + 1
# 发送投票请求给其他节点
for peer in self.peers:
if self._request_vote(peer, self.current_term, len(self.log), self._get_last_log_term()):
votes += 1
# 如果获得多数票，成为leader
if votes > total_nodes // 2:
self._become_leader()
答案解析：
- Raft通过领导者选举、日志复制、安全性保证实现一致性
- 相比Paxos更易理解和实现
- 支持日志压缩、成员变更等高级特性
- 广泛应用于etcd、Consul等分布式系统中
7. 实现分布式锁服务
import time
import uuid
import threading
from typing import Dict, Optional, Set
import redis
from contextlib import contextmanager
class DistributedLock:
"""基于Redis的分布式锁"""
def __init__(self, redis_client, key: str,
timeout: int = 30, retry_delay: float = 0.1):
self.redis = redis_client
self.key = key
self.timeout = timeout
self.retry_delay = retry_delay
self.identifier = str(uuid.uuid4())
def acquire(self, blocking: bool = True, timeout: Optional[float] = None) -> bool:
"""获取锁"""
end_time = time.time() + timeout if timeout else None
while True:
# 使用SET命令的NX和EX选项实现原子性
if self.redis.set(self.key, self.identifier, nx=True, ex=self.timeout):
return True
if not blocking:
return False
if end_time and time.time() > end_time:
return False
time.sleep(self.retry_delay)
def release(self) -> bool:
"""释放锁"""
# 使用Lua脚本确保原子性
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = self.redis.eval(lua_script, 1, self.key, self.identifier)
return result == 1
答案解析：
- 分布式锁确保在分布式环境中的互斥访问
- 使用Redis的SET NX EX命令保证原子性
- 通过唯一标识符防止误释放
- 支持锁续期、超时机制，避免死锁
8. 实现高性能消息队列
import threading
import time
import heapq
from collections import deque
from typing import Any, Callable, Optional, List
from dataclasses import dataclass
from enum import Enum
class MessagePriority(Enum):
LOW = 1
NORMAL = 2
HIGH = 3
URGENT = 4
@dataclass
class Message:
id: str
payload: Any
priority: MessagePriority = MessagePriority.NORMAL
timestamp: float = None
retry_count: int = 0
max_retries: int = 3
def __post_init__(self):
if self.timestamp is None:
self.timestamp = time.time()
class HighPerformanceQueue:
def __init__(self, max_size: int = 10000):
self.max_size = max_size
self.queues = {
MessagePriority.URGENT: deque(),
MessagePriority.HIGH: deque(),
MessagePriority.NORMAL: deque(),
MessagePriority.LOW: deque()
}
self.delayed_messages = [] # 延迟消息堆
self.dead_letter_queue = deque() # 死信队列
self.lock = threading.RLock()
self.not_empty = threading.Condition(self.lock)
self.not_full = threading.Condition(self.lock)
# 统计信息
self.stats = {
'total_messages': 0,
'processed_messages': 0,
'failed_messages': 0,
'dead_letter_count': 0
}
def put(self, message: Message, delay: float = 0) -> bool:
"""添加消息"""
with self.not_full:
# 检查队列是否已满
if self._get_total_size() >= self.max_size:
return False
if delay > 0:
# 延迟消息
execute_time = time.time() + delay
heapq.heappush(self.delayed_messages, (execute_time, message))
else:
# 立即消息
self.queues[message.priority].append(message)
self.stats['total_messages'] += 1
self.not_empty.notify()
return True
def get(self, timeout: Optional[float] = None) -> Optional[Message]:
"""获取消息"""
with self.not_empty:
# 处理延迟消息
self._process_delayed_messages()
# 等待消息
end_time = time.time() + timeout if timeout else None
while self._is_empty() and (end_time is None or time.time() < end_time):
remaining_time = None
if end_time:
remaining_time = end_time - time.time()
if remaining_time <= 0:
break
self.not_empty.wait(remaining_time)
self._process_delayed_messages()
# 按优先级获取消息
for priority in [MessagePriority.URGENT, MessagePriority.HIGH,
MessagePriority.NORMAL, MessagePriority.LOW]:
if self.queues[priority]:
message = self.queues[priority].popleft()
self.stats['processed_messages'] += 1
return message
return None
def ack(self, message: Message) -> None:
"""确认消息处理成功"""
# 在实际实现中，这里会移除消息的确认等待
pass
def nack(self, message: Message, requeue: bool = True) -> None:
"""确认消息处理失败"""
message.retry_count += 1
if message.retry_count <= message.max_retries and requeue:
# 重新入队
self.put(message)
else:
# 加入死信队列
self.dead_letter_queue.append(message)
self.stats['dead_letter_count'] += 1
self.stats['failed_messages'] += 1
def _process_delayed_messages(self) -> None:
"""处理延迟消息"""
current_time = time.time()
while self.delayed_messages and self.delayed_messages[0][0] <= current_time:
_, message = heapq.heappop(self.delayed_messages)
self.queues[message.priority].append(message)
self.stats['total_messages'] += 1
def _get_total_size(self) -> int:
"""获取队列总大小"""
return sum(len(queue) for queue in self.queues.values()) + len(self.delayed_messages)
def _is_empty(self) -> bool:
"""检查队列是否为空"""
return all(len(queue) == 0 for queue in self.queues.values()) and not self.delayed_messages
答案解析：
- 高性能消息队列支持优先级、延迟投递、重试机制
- 使用多级队列和堆结构优化性能
- 支持死信队列处理失败消息
- 线程安全设计，支持多生产者多消费者
9. 实现分布式缓存系统
import hashlib
import json
import time
import threading
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
class CacheConsistency(Enum):
EVENTUAL = "eventual"
STRONG = "strong"
@dataclass
class CacheEntry:
value: Any
timestamp: float
ttl: int
version: int = 1
def is_expired(self) -> bool:
return time.time() - self.timestamp > self.ttl
class DistributedCache:
def __init__(self, node_id: str, nodes: List[str],
consistency: CacheConsistency = CacheConsistency.EVENTUAL):
self.node_id = node_id
self.nodes = nodes
self.consistency = consistency
# 本地缓存
self.local_cache: Dict[str, CacheEntry] = {}
self.lock = threading.RLock()
# 一致性哈希环
self.hash_ring = self._build_hash_ring()
# 版本向量
self.version_vector: Dict[str, int] = {node: 0 for node in nodes}
# 统计信息
self.stats = {
'hits': 0,
'misses': 0,
'evictions': 0,
'replications': 0
}
def _build_hash_ring(self) -> Dict[int, str]:
"""构建一致性哈希环"""
ring = {}
replicas = 100
for node in self.nodes:
for i in range(replicas):
key = f"{node}:{i}"
hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
ring[hash_value] = node
return dict(sorted(ring.items()))
def _get_responsible_nodes(self, key: str, replica_count: int = 3) -> List[str]:
"""获取负责该key的节点"""
hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
# 找到第一个大于等于hash值的节点
sorted_hashes = sorted(self.hash_ring.keys())
index = 0
for i, h in enumerate(sorted_hashes):
if h >= hash_value:
index = i
break
# 获取replica_count个不同的节点
nodes = []
seen = set()
for i in range(len(sorted_hashes)):
current_index = (index + i) % len(sorted_hashes)
node = self.hash_ring[sorted_hashes[current_index]]
if node not in seen:
nodes.append(node)
seen.add(node)
if len(nodes) >= replica_count:
break
return nodes
def get(self, key: str) -> Optional[Any]:
"""获取缓存值"""
with self.lock:
# 检查本地缓存
if key in self.local_cache:
entry = self.local_cache[key]
if not entry.is_expired():
self.stats['hits'] += 1
return entry.value
else:
del self.local_cache[key]
self.stats['misses'] += 1
# 如果是强一致性，从主节点获取
if self.consistency == CacheConsistency.STRONG:
return self._get_from_primary(key)
# 最终一致性，从任意副本获取
return self._get_from_replica(key)
def put(self, key: str, value: Any, ttl: int = 3600) -> bool:
"""设置缓存值"""
with self.lock:
timestamp = time.time()
entry = CacheEntry(value, timestamp, ttl, self.version_vector[self.node_id] + 1)
# 更新本地缓存
self.local_cache[key] = entry
self.version_vector[self.node_id] += 1
# 异步复制到其他节点
responsible_nodes = self._get_responsible_nodes(key)
for node in responsible_nodes:
if node != self.node_id:
self._replicate_to_node(node, key, entry)
self.stats['replications'] += 1
return True
def delete(self, key: str) -> bool:
"""删除缓存值"""
with self.lock:
if key in self.local_cache:
del self.local_cache[key]
# 通知其他节点删除
responsible_nodes = self._get_responsible_nodes(key)
for node in responsible_nodes:
if node != self.node_id:
self._delete_from_node(node, key)
return True
def _get_from_primary(self, key: str) -> Optional[Any]:
"""从主节点获取数据"""
responsible_nodes = self._get_responsible_nodes(key, 1)
if responsible_nodes and responsible_nodes[0] != self.node_id:
# 模拟从其他节点获取
return None
return None
def _get_from_replica(self, key: str) -> Optional[Any]:
"""从副本节点获取数据"""
responsible_nodes = self._get_responsible_nodes(key)
for node in responsible_nodes:
if node != self.node_id:
# 模拟从其他节点获取
pass
return None
def _replicate_to_node(self, node: str, key: str, entry: CacheEntry) -> None:
"""复制数据到指定节点"""
# 在实际实现中，这里会通过网络调用
pass
def _delete_from_node(self, node: str, key: str) -> None:
"""从指定节点删除数据"""
# 在实际实现中，这里会通过网络调用
pass
def cleanup_expired(self) -> int:
"""清理过期条目"""
with self.lock:
expired_keys = []
current_time = time.time()
for key, entry in self.local_cache.items():
if entry.is_expired():
expired_keys.append(key)
for key in expired_keys:
del self.local_cache[key]
self.stats['evictions'] += 1
return len(expired_keys)
def get_stats(self) -> Dict[str, Any]:
"""获取统计信息"""
with self.lock:
total_requests = self.stats['hits'] + self.stats['misses']
hit_rate = self.stats['hits'] / total_requests if total_requests > 0 else 0
return {
**self.stats,
'hit_rate': hit_rate,
'cache_size': len(self.local_cache),
'version_vector': self.version_vector.copy()
}
答案解析：
- 分布式缓存系统支持多节点数据一致性
- 使用一致性哈希实现数据分布
- 支持强一致性和最终一致性两种模式
- 包含过期清理、统计监控等完整功能
10. 实现高性能连接池
import threading
import time
import queue
import socket
import ssl
from typing import Optional, Dict, Any, List
from contextlib import contextmanager
from dataclasses import dataclass
@dataclass
class ConnectionConfig:
host: str
port: int
timeout: float = 30.0
max_retries: int = 3
use_ssl: bool = False
ssl_context: Optional[ssl.SSLContext] = None
class PooledConnection:
def __init__(self, connection: socket.socket, pool: 'ConnectionPool',
created_time: float, last_used: float):
self.connection = connection
self.pool = pool
self.created_time = created_time
self.last_used = last_used
self.in_use = False
self.is_valid = True
def __enter__(self):
self.in_use = True
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb):
self.in_use = False
self.last_used = time.time()
self.pool._return_connection(self)
def close(self):
"""关闭连接"""
if self.is_valid:
try:
self.connection.close()
except:
pass
finally:
self.is_valid = False
class ConnectionPool:
def __init__(self, config: ConnectionConfig,
min_connections: int = 5,
max_connections: int = 50,
max_idle_time: float = 300.0,
connection_lifetime: float = 3600.0):
self.config = config
self.min_connections = min_connections
self.max_connections = max_connections
self.max_idle_time = max_idle_time
self.connection_lifetime = connection_lifetime
self._pool = queue.Queue(maxsize=max_connections)
self._active_connections: Dict[int, PooledConnection] = {}
self._lock = threading.RLock()
# 统计信息
self.stats = {
'created': 0,
'destroyed': 0,
'borrowed': 0,
'returned': 0,
'active_count': 0,
'idle_count': 0
}
# 初始化最小连接数
self._initialize_pool()
# 启动清理线程
self._cleanup_thread = threading.Thread(target=self._cleanup_worker, daemon=True)
self._cleanup_thread.start()
def _initialize_pool(self) -> None:
"""初始化连接池"""
for _ in range(self.min_connections):
try:
conn = self._create_connection()
if conn:
self._pool.put(conn)
except Exception as e:
print(f"初始化连接失败: {e}")
def _create_connection(self) -> Optional[PooledConnection]:
"""创建新连接"""
try:
# 创建socket连接
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(self.config.timeout)
# SSL包装
if self.config.use_ssl:
if self.config.ssl_context:
sock = self.config.ssl_context.wrap_socket(sock, server_hostname=self.config.host)
else:
sock = ssl.create_default_context().wrap_socket(sock, server_hostname=self.config.host)
# 连接
sock.connect((self.config.host, self.config.port))
current_time = time.time()
pooled_conn = PooledConnection(sock, self, current_time, current_time)
with self._lock:
self._active_connections[id(sock)] = pooled_conn
self.stats['created'] += 1
self.stats['active_count'] += 1
return pooled_conn
except Exception as e:
print(f"创建连接失败: {e}")
return None
def get_connection(self, timeout: Optional[float] = None) -> Optional[PooledConnection]:
"""获取连接"""
start_time = time.time()
while True:
# 尝试从池中获取连接
try:
conn = self._pool.get_nowait()
# 检查连接是否有效
if self._is_connection_valid(conn):
with self._lock:
self.stats['borrowed'] += 1
self.stats['idle_count'] -= 1
return conn
else:
# 连接无效，销毁并继续
self._destroy_connection(conn)
except queue.Empty:
pass
# 池中没有可用连接，尝试创建新连接
with self._lock:
current_count = len(self._active_connections)
if current_count < self.max_connections:
new_conn = self._create_connection()
if new_conn:
self.stats['borrowed'] += 1
return new_conn
# 检查超时
if timeout and (time.time() - start_time) > timeout:
return None
# 等待连接返回
try:
conn = self._pool.get(timeout=0.1)
if self._is_connection_valid(conn):
with self._lock:
self.stats['borrowed'] += 1
self.stats['idle_count'] -= 1
return conn
else:
self._destroy_connection(conn)
except queue.Empty:
continue
def _return_connection(self, conn: PooledConnection) -> None:
"""归还连接"""
if not conn.is_valid:
self._destroy_connection(conn)
return
try:
self._pool.put_nowait(conn)
with self._lock:
self.stats['returned'] += 1
self.stats['idle_count'] += 1
except queue.Full:
# 池已满，销毁连接
self._destroy_connection(conn)
def _is_connection_valid(self, conn: PooledConnection) -> bool:
"""检查连接是否有效"""
if not conn or not conn.is_valid:
return False
current_time = time.time()
# 检查连接是否超时
if current_time - conn.last_used > self.max_idle_time:
return False
# 检查连接生命周期
if current_time - conn.created_time > self.connection_lifetime:
return False
# 简单的ping检查
try:
conn.connection.send(b'PING\r\n')
return True
except:
return False
def _destroy_connection(self, conn: PooledConnection) -> None:
"""销毁连接"""
conn.close()
with self._lock:
conn_id = id(conn.connection)
if conn_id in self._active_connections:
del self._active_connections[conn_id]
self.stats['destroyed'] += 1
self.stats['active_count'] -= 1
if not conn.in_use:
self.stats['idle_count'] -= 1
def _cleanup_worker(self) -> None:
"""清理工作线程"""
while True:
try:
time.sleep(60) # 每分钟清理一次
self._cleanup_idle_connections()
except Exception as e:
print(f"清理线程异常: {e}")
def _cleanup_idle_connections(self) -> None:
"""清理空闲连接"""
current_time = time.time()
connections_to_destroy = []
# 检查池中的连接
temp_connections = []
while not self._pool.empty():
try:
conn = self._pool.get_nowait()
if current_time - conn.last_used > self.max_idle_time:
connections_to_destroy.append(conn)
else:
temp_connections.append(conn)
except queue.Empty:
break
# 将有效连接放回池中
for conn in temp_connections:
try:
self._pool.put_nowait(conn)
except queue.Full:
connections_to_destroy.append(conn)
# 销毁过期连接
for conn in connections_to_destroy:
self._destroy_connection(conn)
@contextmanager
def connection(self, timeout: Optional[float] = None):
"""连接上下文管理器"""
conn = self.get_connection(timeout)
if conn is None:
raise ConnectionError("无法获取连接")
try:
with conn as raw_conn:
yield raw_conn
except Exception as e:
# 连接出错，标记为无效
conn.is_valid = False
raise
def get_stats(self) -> Dict[str, Any]:
"""获取连接池统计信息"""
with self._lock:
return {
**self.stats,
'pool_size': self._pool.qsize(),
'total_connections': len(self._active_connections)
}
def close(self) -> None:
"""关闭连接池"""
# 销毁所有连接
connections_to_destroy = []
# 清空池
while not self._pool.empty():
try:
conn = self._pool.get_nowait()
connections_to_destroy.append(conn)
except queue.Empty:
break
# 销毁所有连接
for conn in connections_to_destroy:
self._destroy_connection(conn)
# 销毁活跃连接
with self._lock:
for conn in list(self._active_connections.values()):
self._destroy_connection(conn)
答案解析：
- 高性能连接池支持动态扩缩容、连接复用、健康检查
- 使用队列管理空闲连接，支持并发访问
- 包含连接生命周期管理、超时清理机制
- 提供统计监控和上下文管理器，使用便捷
11. 实现快速排序并优化其性能
import random
def quick_sort_optimized(arr):
"""优化的快速排序"""
if len(arr) <= 1:
return arr
# 三数取中法选择基准
first, middle, last = arr[0], arr[len(arr)//2], arr[-1]
pivot = sorted([first, middle, last])[1]
left = [x for x in arr if x < pivot]
middle = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
return quick_sort_optimized(left) + middle + quick_sort_optimized(right)
def quick_sort_inplace(arr, low=0, high=None):
"""原地快速排序"""
if high is None:
high = len(arr) - 1
if low < high:
# 小数组使用插入排序
if high - low < 10:
insertion_sort(arr, low, high)
return
# 分区
pi = partition(arr, low, high)
# 递归排序
quick_sort_inplace(arr, low, pi - 1)
quick_sort_inplace(arr, pi + 1, high)
def partition(arr, low, high):
"""分区函数"""
# 随机选择基准
pivot_index = random.randint(low, high)
arr[pivot_index], arr[high] = arr[high], arr[pivot_index]
pivot = arr[high]
i = low - 1
for j in range(low, high):
if arr[j] <= pivot:
i += 1
arr[i], arr[j] = arr[j], arr[i]
arr[i + 1], arr[high] = arr[high], arr[i + 1]
return i + 1
def insertion_sort(arr, low, high):
"""插入排序"""
for i in range(low + 1, high + 1):
key = arr[i]
j = i - 1
while j >= low and arr[j] > key:
arr[j + 1] = arr[j]
j -= 1
arr[j + 1] = key
答案解析：
- 使用三数取中法避免最坏情况
- 小数组使用插入排序减少递归开销
- 随机化基准选择提高性能
- 平均时间复杂度O(n log n)，最坏情况O(n²)
12. 实现一个高效的内存池
import ctypes
import threading
from typing import Dict, List, Optional
class MemoryPool:
def __init__(self, block_size: int = 1024, num_blocks: int = 100):
"""
内存池
block_size: 每个块的大小
num_blocks: 初始块数量
"""
self.block_size = block_size
self.num_blocks = num_blocks
self.free_blocks = []
self.allocated_blocks = set()
self.lock = threading.Lock()
# 预分配内存块
self._allocate_blocks()
def _allocate_blocks(self):
"""预分配内存块"""
for _ in range(self.num_blocks):
block = ctypes.create_string_buffer(self.block_size)
self.free_blocks.append(block)
def allocate(self) -> Optional[ctypes.c_char_p]:
"""分配内存块"""
with self.lock:
if not self.free_blocks:
# 动态扩展
try:
block = ctypes.create_string_buffer(self.block_size)
self.allocated_blocks.add(id(block))
return ctypes.cast(block, ctypes.c_char_p)
except:
return None
block = self.free_blocks.pop()
self.allocated_blocks.add(id(block))
return ctypes.cast(block, ctypes.c_char_p)
def deallocate(self, block: ctypes.c_char_p) -> bool:
"""释放内存块"""
with self.lock:
block_id = ctypes.addressof(block.contents)
if block_id not in self.allocated_blocks:
return False
# 将块重新加入空闲列表
block_obj = ctypes.cast(block, ctypes.c_char_p)
self.free_blocks.append(block_obj)
self.allocated_blocks.remove(block_id)
return True
def get_stats(self) -> Dict:
"""获取统计信息"""
with self.lock:
return {
'free_blocks': len(self.free_blocks),
'allocated_blocks': len(self.allocated_blocks),
'block_size': self.block_size
}
答案解析：
- 预分配内存块减少系统调用开销
- 支持动态扩展和线程安全
- 使用RAII管理内存生命周期
- 适用于高频内存分配场景
13. 实现红黑树的插入操作
from enum import Enum
class Color(Enum):
RED = 1
BLACK = 2
class RBNode:
def __init__(self, key, value=None):
self.key = key
self.value = value
self.left = None
self.right = None
self.parent = None
self.color = Color.RED
class RedBlackTree:
def __init__(self):
self.NIL = RBNode(None) # 哨兵节点
self.NIL.color = Color.BLACK
self.root = self.NIL
def insert(self, key, value=None):
"""插入节点"""
node = RBNode(key, value)
node.left = self.NIL
node.right = self.NIL
parent = None
current = self.root
while current != self.NIL:
parent = current
if node.key < current.key:
current = current.left
else:
current = current.right
node.parent = parent
if parent is None:
self.root = node
elif node.key < parent.key:
parent.left = node
else:
parent.right = node
self._fix_insert(node)
def _fix_insert(self, node):
"""修复插入后的红黑树性质"""
while node.parent and node.parent.color == Color.RED:
if node.parent == node.parent.parent.left:
uncle = node.parent.parent.right
if uncle.color == Color.RED:
# 情况1：叔叔是红色
node.parent.color = Color.BLACK
uncle.color = Color.BLACK
node.parent.parent.color = Color.RED
node = node.parent.parent
else:
if node == node.parent.right:
# 情况2：叔叔是黑色，节点是右孩子
node = node.parent
self._left_rotate(node)
# 情况3：叔叔是黑色，节点是左孩子
node.parent.color = Color.BLACK
node.parent.parent.color = Color.RED
self._right_rotate(node.parent.parent)
else:
# 对称情况
uncle = node.parent.parent.left
if uncle.color == Color.RED:
node.parent.color = Color.BLACK
uncle.color = Color.BLACK
node.parent.parent.color = Color.RED
node = node.parent.parent
else:
if node == node.parent.left:
node = node.parent
self._right_rotate(node)
node.parent.color = Color.BLACK
node.parent.parent.color = Color.RED
self._left_rotate(node.parent.parent)
self.root.color = Color.BLACK
def _left_rotate(self, x):
"""左旋转"""
y = x.right
x.right = y.left
if y.left != self.NIL:
y.left.parent = x
y.parent = x.parent
if x.parent is None:
self.root = y
elif x == x.parent.left:
x.parent.left = y
else:
x.parent.right = y
y.left = x
x.parent = y
def _right_rotate(self, y):
"""右旋转"""
x = y.left
y.left = x.right
if x.right != self.NIL:
x.right.parent = y
x.parent = y.parent
if y.parent is None:
self.root = x
elif y == y.parent.right:
y.parent.right = x
else:
y.parent.left = x
x.right = y
y.parent = x
答案解析：
- 红黑树保持平衡，确保O(log n)的操作复杂度
- 通过颜色和旋转操作维护树的平衡性质
- 插入后通过三种情况修复红黑树性质
- 支持动态大小统计
14. 实现Dijkstra最短路径算法
import heapq
from typing import Dict, List, Tuple, Set, Optional
class Graph:
def __init__(self):
self.adjacency_list = {}
def add_edge(self, u: str, v: str, weight: float):
"""添加边"""
if u not in self.adjacency_list:
self.adjacency_list[u] = []
if v not in self.adjacency_list:
self.adjacency_list[v] = []
self.adjacency_list[u].append((v, weight))
# 无向图需要添加反向边
self.adjacency_list[v].append((u, weight))
def dijkstra(self, start: str) -> Tuple[Dict[str, float], Dict[str, List[str]]]:
"""
Dijkstra算法计算最短路径
返回：(距离字典, 路径字典)
"""
distances = {node: float('inf') for node in self.adjacency_list}
distances[start] = 0
previous = {node: None for node in self.adjacency_list}
# 优先队列：(距离, 节点)
pq = [(0, start)]
visited = set()
while pq:
current_distance, current_node = heapq.heappop(pq)
if current_node in visited:
continue
visited.add(current_node)
# 更新邻居节点的距离
for neighbor, weight in self.adjacency_list[current_node]:
if neighbor in visited:
continue
distance = current_distance + weight
if distance < distances[neighbor]:
distances[neighbor] = distance
previous[neighbor] = current_node
heapq.heappush(pq, (distance, neighbor))
# 构建路径
paths = {}
for node in self.adjacency_list:
if distances[node] == float('inf'):
paths[node] = (float('inf'), [])
else:
path = []
current = node
while current is not None:
path.append(current)
current = previous[current]
paths[node] = (distances[node], list(reversed(path)))
return distances, paths
def shortest_path(self, start: str, end: str) -> Tuple[float, List[str]]:
"""获取两点间最短路径"""
distances, paths = self.dijkstra(start)
return paths.get(end, (float('inf'), []))
答案解析：
- Dijkstra算法使用贪心策略计算单源最短路径
- 使用优先队列优化性能，时间复杂度O((V+E)logV)
- 支持路径回溯和距离计算
- 不适用于负权边
15. 实现一个高效的哈希表
import hashlib
from typing import Any, Dict, List, Optional, Tuple
class HashTable:
def __init__(self, capacity: int = 16, load_factor: float = 0.75):
"""
哈希表
capacity: 初始容量
load_factor: 负载因子
"""
self.capacity = capacity
self.size = 0
self.load_factor = load_factor
self.buckets = [[] for _ in range(capacity)] # 使用链地址法
def _hash(self, key: Any) -> int:
"""计算哈希值"""
if isinstance(key, int):
return key % self.capacity
elif isinstance(key, str):
# 使用MD5确保字符串哈希分布均匀
hash_obj = hashlib.md5(key.encode())
return int(hash_obj.hexdigest(), 16) % self.capacity
else:
return hash(key) % self.capacity
def _resize(self):
"""扩容"""
new_capacity = self.capacity * 2
new_buckets = [[] for _ in range(new_capacity)]
# 重新哈希所有元素
for bucket in self.buckets:
for key, value in bucket:
new_hash = hash(key) % new_capacity
new_buckets[new_hash].append((key, value))
self.capacity = new_capacity
self.buckets = new_buckets
def put(self, key: Any, value: Any) -> None:
"""设置键值对"""
if self.size / self.capacity > self.load_factor:
self._resize()
hash_value = self._hash(key)
bucket = self.buckets[hash_value]
# 检查key是否已存在
for i, (k, v) in enumerate(bucket):
if k == key:
bucket[i] = (key, value)
return
# 添加新键值对
bucket.append((key, value))
self.size += 1
def get(self, key: Any) -> Optional[Any]:
"""获取值"""
hash_value = self._hash(key)
bucket = self.buckets[hash_value]
for k, v in bucket:
if k == key:
return v
return None
def remove(self, key: Any) -> bool:
"""删除键值对"""
hash_value = self._hash(key)
bucket = self.buckets[hash_value]
for i, (k, v) in enumerate(bucket):
if k == key:
del bucket[i]
self.size -= 1
return True
return False
def contains_key(self, key: Any) -> bool:
"""检查键是否存在"""
return self.get(key) is not None
def keys(self) -> List[Any]:
"""获取所有键"""
keys = []
for bucket in self.buckets:
keys.extend(k for k, v in bucket)
return keys
def values(self) -> List[Any]:
"""获取所有值"""
values = []
for bucket in self.buckets:
values.extend(v for k, v in bucket)
return values
def items(self) -> List[Tuple[Any, Any]]:
"""获取所有键值对"""
items = []
for bucket in self.buckets:
items.extend(bucket)
return items
def __len__(self) -> int:
"""返回元素数量"""
return self.size
def __str__(self) -> str:
"""字符串表示"""
items = []
for bucket in self.buckets:
items.extend(f"{k}: {v}" for k, v in bucket)
return "{" + ", ".join(items) + "}"
答案解析：
- 使用链地址法处理哈希冲突
- 动态扩容机制保持性能
- 支持任意类型的键值对
- 平均时间复杂度O(1)，最坏情况O(n)
编程语言 (10题)
16. 解释Java内存模型（JMM）和volatile关键字
// Java内存模型示例
public class JMMExample {
private volatile boolean flag = false;
private int counter = 0;
public void writer() {
counter = 42; // 1
flag = true; // 2 - volatile写
}
public void reader() {
if (flag) { // 3 - volatile读
int value = counter; // 4
System.out.println("Counter: " + value);
}
}
}
// happens-before规则示例
public class HappensBeforeExample {
private int value = 0;
private volatile boolean ready = false;
// 线程A执行
public void setValue(int newValue) {
value = newValue; // 1
ready = true; // 2 - volatile写，建立happens-before关系
}
// 线程B执行
public int getValue() {
if (ready) { // 3 - volatile读
return value; // 4 - 能看到线程A的所有写操作
}
return 0;
}
}
答案解析：
- JMM定义了线程和主内存之间的抽象关系
- volatile保证可见性和有序性，但不保证原子性
- happens-before规则定义了操作间的偏序关系
- volatile写之前的所有操作对volatile读之后的操作可见
17. 实现线程安全的单例模式（双重检查锁定）
public class Singleton {
// volatile确保多线程环境下的可见性和有序性
private static volatile Singleton instance;
// 私有构造函数防止外部实例化
private Singleton() {
// 防止反射攻击
if (instance != null) {
throw new IllegalStateException("Singleton already initialized");
}
}
// 双重检查锁定
public static Singleton getInstance() {
if (instance == null) { // 第一次检查，避免不必要的同步
synchronized (Singleton.class) {
if (instance == null) { // 第二次检查，确保线程安全
instance = new Singleton();
}
}
}
return instance;
}
// 防止序列化破坏单例
protected Object readResolve() {
return getInstance();
}
}
// 枚举单例（推荐方式）
public enum EnumSingleton {
INSTANCE;
public void doSomething() {
System.out.println("Singleton method called");
}
}
// 静态内部类单例
public class StaticInnerSingleton {
private StaticInnerSingleton() {}
private static class SingletonHolder {
private static final StaticInnerSingleton INSTANCE =
new StaticInnerSingleton();
}
public static StaticInnerSingleton getInstance() {
return SingletonHolder.INSTANCE;
}
}
答案解析：
- 双重检查锁定减少同步开销，提高性能
- volatile防止指令重排序，确保安全发布
- 枚举单例是最安全、简洁的实现方式
- 静态内部类利用类加载机制保证线程安全
18. 解释Python的GIL及其影响
import threading
import time
import multiprocessing
# CPU密集型任务 - 受GIL影响
def cpu_bound_task(n):
"""CPU密集型任务"""
count = 0
for i in range(n):
count += i * i
return count
# I/O密集型任务 - 不受GIL影响
def io_bound_task():
"""I/O密集型任务"""
time.sleep(1) # 模拟I/O操作
return "completed"
# 多线程测试（受GIL限制）
def test_multithreading():
start_time = time.time()
threads = []
for i in range(4):
t = threading.Thread(target=cpu_bound_task, args=(10000000,))
threads.append(t)
t.start()
for t in threads:
t.join()
end_time = time.time()
print(f"多线程耗时: {end_time - start_time:.2f}秒")
# 多进程测试（绕过GIL）
def test_multiprocessing():
start_time = time.time()
processes = []
for i in range(4):
p = multiprocessing.Process(target=cpu_bound_task, args=(10000000,))
processes.append(p)
p.start()
for p in processes:
p.join()
end_time = time.time()
print(f"多进程耗时: {end_time - start_time:.2f}秒")
# 使用C扩展绕过GIL
import cffi
ffi = cffi.FFI()
ffi.cdef("""
long long cpu_bound_task_c(long long n);
""")
# 编译C代码
# long long cpu_bound_task_c(long long n) {
# long long count = 0;
# for (long long i = 0; i < n; i++) {
# count += i * i;
# }
# return count;
# }
# 加载编译的库
try:
C = ffi.dlopen("./cpu_task.so")
def cpu_task_c(n):
return C.cpu_bound_task_c(n)
def test_c_extension():
start_time = time.time()
threads = []
for i in range(4):
t = threading.Thread(target=cpu_task_c, args=(10000000,))
threads.append(t)
t.start()
for t in threads:
t.join()
end_time = time.time()
print(f"C扩展多线程耗时: {end_time - start_time:.2f}秒")
except:
print("C扩展库未找到")
if __name__ == "__main__":
print("=== GIL影响测试 ===")
test_multithreading()
test_multiprocessing()
try:
test_c_extension()
except:
pass
答案解析：
- GIL是Python解释器的线程安全机制，同一时间只允许一个线程执行Python字节码
- GIL对I/O密集型任务影响较小，对CPU密集型任务影响较大
- 多进程可以绕过GIL限制，充分利用多核CPU
- C扩展可以在执行时释放GIL，提高并发性能
19. 实现一个简单的协程调度器
import types
import time
from collections import deque
class Task:
def __init__(self, coro):
self.coro = coro
self.send_value = None
self.finished = False
def step(self):
"""执行一步"""
try:
if self.send_value is not None:
result = self.coro.send(self.send_value)
self.send_value = None
else:
result = next(self.coro)
return result
except StopIteration:
self.finished = True
return None
except Exception as e:
self.finished = True
raise e
class Scheduler:
def __init__(self):
self.ready = deque() # 就绪队列
self.sleeping = [] # 睡眠任务
self.current = None # 当前执行的任务
def new_task(self, coro):
"""创建新任务"""
task = Task(coro)
self.ready.append(task)
return task
def sleep(self, delay):
"""让当前任务睡眠"""
self.current.send_value = ('sleep', delay)
yield
def schedule(self):
"""调度运行"""
while self.ready or self.sleeping:
# 处理睡眠任务
current_time = time.time()
while self.sleeping and self.sleeping[0][0] <= current_time:
_, task = heapq.heappop(self.sleeping)
self.ready.append(task)
if not self.ready:
# 没有就绪任务，等待
time.sleep(0.1)
continue
# 获取下一个任务
task = self.ready.popleft()
self.current = task
# 执行任务
result = task.step()
if task.finished:
continue
# 处理任务结果
if isinstance(result, tuple) and result[0] == 'sleep':
delay = result[1]
heapq.heappush(self.sleeping, (current_time + delay, task))
else:
# 任务继续执行
self.ready.append(task)
# 使用示例
def countdown(name, n):
"""倒计时协程"""
while n > 0:
print(f"{name}: {n}")
yield scheduler.sleep(1)
n -= 1
print(f"{name}: 完成!")
def producer():
"""生产者协程"""
for i in range(5):
print(f"生产: {i}")
yield scheduler.sleep(0.5)
print("生产完成")
def consumer():
"""消费者协程"""
for i in range(5):
print(f"消费: {i}")
yield scheduler.sleep(0.8)
print("消费完成")
if __name__ == "__main__":
import heapq
scheduler = Scheduler()
# 创建任务
scheduler.new_task(countdown("Timer1", 5))
scheduler.new_task(countdown("Timer2", 3))
scheduler.new_task(producer())
scheduler.new_task(consumer())
# 开始调度
scheduler.schedule()
答案解析：
- 协程是用户态的轻量级线程，切换开销小
- 使用生成器实现简单的协程调度
- 支持睡眠、任务切换等基本功能
- 可以扩展为更复杂的异步框架
20. 解释C++的智能指针和RAII
#include <iostream>
#include <memory>
#include <vector>
#include <string>
// RAII资源管理类
class FileHandler {
private:
FILE* file;
public:
FileHandler(const char* filename, const char* mode) {
file = fopen(filename, mode);
if (!file) {
throw std::runtime_error("Failed to open file");
}
std::cout << "File opened" << std::endl;
}
~FileHandler() {
if (file) {
fclose(file);
std::cout << "File closed" << std::endl;
}
}
// 禁止拷贝
FileHandler(const FileHandler&) = delete;
FileHandler& operator=(const FileHandler&) = delete;
// 允许移动
FileHandler(FileHandler&& other) noexcept : file(other.file) {
other.file = nullptr;
}
FileHandler& operator=(FileHandler&& other) noexcept {
if (this != &other) {
if (file) fclose(file);
file = other.file;
other.file = nullptr;
}
return *this;
}
void write(const std::string& data) {
if (file) {
fprintf(file, "%s\n", data.c_str());
}
}
};
// 智能指针示例
class Resource {
private:
std::string name;
public:
Resource(const std::string& n) : name(n) {
std::cout << "Resource " << name << " created" << std::endl;
}
~Resource() {
std::cout << "Resource " << name << " destroyed" << std::endl;
}
void use() {
std::cout << "Using resource " << name << std::endl;
}
};
void demonstrate_smart_pointers() {
// unique_ptr - 独占所有权
{
std::unique_ptr<Resource> ptr1 = std::make_unique<Resource>("A");
ptr1->use();
// 转移所有权
std::unique_ptr<Resource> ptr2 = std::move(ptr1);
ptr2->use();
// ptr1现在为空
} // ptr2自动释放
// shared_ptr - 共享所有权
{
std::shared_ptr<Resource> ptr1 = std::make_shared<Resource>("B");
{
std::shared_ptr<Resource> ptr2 = ptr1; // 引用计数+1
std::cout << "Use count: " << ptr1.use_count() << std::endl;
ptr2->use();
} // ptr2释放，引用计数-1
std::cout << "Use count: " << ptr1.use_count() << std::endl;
} // ptr1释放，资源被销毁
// weak_ptr - 弱引用，解决循环引用
{
std::shared_ptr<Resource> ptr1 = std::make_shared<Resource>("C");
std::weak_ptr<Resource> weak_ptr = ptr1;
std::cout << "Use count: " << ptr1.use_count() << std::endl;
if (auto shared_ptr = weak_ptr.lock()) { // 尝试获取shared_ptr
shared_ptr->use();
}
std::cout << "Use count: " << ptr1.use_count() << std::endl;
}
}
// 自定义删除器
void custom_deleter() {
auto custom_delete = [](int* p) {
std::cout << "Custom deleter called" << std::endl;
delete p;
};
std::unique_ptr<int, decltype(custom_delete)> ptr(new int(42), custom_delete);
}
int main() {
std::cout << "=== RAII Demo ===" << std::endl;
{
FileHandler file("test.txt", "w");
file.write("Hello, RAII!");
} // 文件自动关闭
std::cout << "\n=== Smart Pointers Demo ===" << std::endl;
demonstrate_smart_pointers();
std::cout << "\n=== Custom Deleter Demo ===" << std::endl;
custom_deleter();
return 0;
}
答案解析：
- RAII（Resource Acquisition Is Initialization）通过对象生命周期管理资源
- unique_ptr提供独占所有权，不可拷贝但可移动
- shared_ptr提供共享所有权，使用引用计数管理生命周期
- weak_ptr解决循环引用问题，不增加引用计数
100. 请解释C++中的移动语义和完美转发
#include <utility>
#include <iostream>
#include <string>
class Resource {
private:
std::string data;
size_t* ref_count;
public:
Resource(const std::string& str) : data(str), ref_count(new size_t(1)) {
std::cout << "Constructor: " << data << std::endl;
}
// 拷贝构造函数
Resource(const Resource& other) : data(other.data), ref_count(other.ref_count) {
(*ref_count)++;
std::cout << "Copy Constructor: " << data << std::endl;
}
// 移动构造函数
Resource(Resource&& other) noexcept
: data(std::move(other.data)), ref_count(other.ref_count) {
other.ref_count = nullptr;
std::cout << "Move Constructor: " << data << std::endl;
}
// 拷贝赋值运算符
Resource& operator=(const Resource& other) {
if (this != &other) {
cleanup();
data = other.data;
ref_count = other.ref_count;
(*ref_count)++;
std::cout << "Copy Assignment: " << data << std::endl;
}
return *this;
}
// 移动赋值运算符
Resource& operator=(Resource&& other) noexcept {
if (this != &other) {
cleanup();
data = std::move(other.data);
ref_count = other.ref_count;
other.ref_count = nullptr;
std::cout << "Move Assignment: " << data << std::endl;
}
return *this;
}
~Resource() {
cleanup();
}
private:
void cleanup() {
if (ref_count && --(*ref_count) == 0) {
delete ref_count;
std::cout << "Destructor: " << data << std::endl;
}
}
};
// 完美转发示例
template<typename T>
void perfect_forward(T&& arg) {
process(std::forward<T>(arg));
}
void process(const Resource& res) {
std::cout << "Processing lvalue reference" << std::endl;
}
void process(Resource&& res) {
std::cout << "Processing rvalue reference" << std::endl;
}
// 工厂函数示例
Resource create_resource(const std::string& name) {
return Resource(name);
}
int main() {
Resource res1("Hello");
Resource res2 = res1; // 拷贝构造
Resource res3 = std::move(res1); // 移动构造
Resource res4 = create_resource("World"); // 移动构造
res2 = res3; // 拷贝赋值
res4 = Resource("Temporary"); // 移动赋值
perfect_forward(res2); // 传递左值
perfect_forward(Resource("Temp")); // 传递右值
return 0;
}
核心概念：
- 移动语义：将资源所有权从一个对象转移到另一个对象，避免不必要的拷贝
- 右值引用：使用&&绑定到右值，延长临时对象生命周期
- 完美转发：使用std::forward保持参数的值类别（左值/右值）
- RAII：资源获取即初始化，自动管理资源生命周期
🎯 总结
本文件涵盖了阿里、百度等互联网大厂的100道经典面试题，包括：
✅ 完成内容
- 编程基础 (30题)：数据结构、算法、设计模式
- 系统设计 (25题)：分布式系统、数据库、搜索引擎
- 公司特色 (45题)：阿里电商架构、百度AI技术
🚀 技术亮点
- 完整代码实现：每道题都包含可运行的代码示例
- 深度技术解析：不仅讲"是什么"，更讲"为什么"和"怎么做"
- 实战导向：结合实际业务场景，注重工程实践
- 前沿技术：涵盖微服务、云原生、AI等热门技术栈
📚 学习建议
- 循序渐进：从基础数据结构开始，逐步深入到系统设计
- 动手实践：每道题都要亲自编码实现，理解底层原理
- 举一反三：思考同一问题的不同解决方案和权衡
- 持续更新：关注技术发展趋势，学习最新架构模式
🎯 面试技巧
- 思路清晰：先分析问题，再设计方案，最后实现代码
- 沟通表达：及时向面试官说明思路和考虑因素
- 代码规范：注意命名规范、错误处理、边界条件
- 性能意识：考虑时间复杂度、空间复杂度、并发安全
📝 文档说明
- 题目总数：100道
- 代码行数：约3000行
- 涵盖技术栈：Java、Python、C++、Go、分布式系统、数据库、AI
- 难度等级：中级到高级，适合3-8年经验工程师
🔄 持续更新
本文档会根据最新的技术趋势和面试要求持续更新，建议定期查看最新版本。
更多推荐
20
0- 0
已为社区贡献4条内容
所有评论(0)