package main
import "fmt"
func f(a [3]int) {
    a[1] = 123
}
func g(s []int){
    s[1]=123
}
func main() {
    arr := [3]int{1, 2, 3}
    fmt.Println(arr)
    f(arr)
    fmt.Println(arr)
    var slice1 []int
    slice1 = []int{1, 2, 3}
    fmt.Println(slice1)
    g(slice1)
    fmt.Println(slice1)
}

a=[1,2,3]
def f(b):
    b[1]=123
print(a)
f(a)
print(a)

[1, 2, 3]
[1, 123, 3]

数组内存分布

var arr [5]int{1, 2, 3, 4, 5}

切片内存分布

slice := []int{1, 2, 3, 4, 5}

Slice Header (24 bytes on 64-bit):
┌─────────────────┐
│ array: 0x100 │ ← 指针（8 bytes）
├─────────────────┤
│ len: 5 │ ← 长度（8 bytes）
├─────────────────┤
│ cap: 5 │ ← 容量（8 bytes）
└─────────────────┘
↓ 指向
底层数组:
┌──────┬──────┬──────┬──────┬──────┐
│ 1 │ 2 │ 3 │ 4 │ 5 │
└──────┴──────┴──────┴──────┴──────┘

Python: Thread 1 [获取GIL] → 执行 → [释放GIL] → Thread 2 [等待] → [获取GIL] → ...
Java:   Thread 1 ████████ 真正并行执行
        Thread 2 ████████ 真正并行执行
        Thread 3 ████████ 真正并行执行

 import threading
import time
import requests
from collections import deque
class RateLimiter:
    """API 速率限制器 - 基于滑动窗口算法"""
    def __init__(self, max_requests=10, time_window=60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.semaphore = threading.Semaphore(max_requests)
        self.request_times = deque()
        self.lock = threading.Lock()
        self._shutdown = False
    def _clean_old_requests(self):
        """清理超出时间窗口的请求记录"""
        now = time.time()
        cutoff = now - self.time_window
        while self.request_times and self.request_times[0] < cutoff:
            self.request_times.popleft()
    def _wait_for_slot(self):
        """等待可用的请求槽位"""
        while not self._shutdown:
            with self.lock:
                self._clean_old_requests()
                if len(self.request_times) < self.max_requests:
                    self.request_times.append(time.time())
                    return True
            # 计算需要等待的时间
            with self.lock:
                if self.request_times:
                    oldest = self.request_times[0]
                    wait_time = self.time_window - (time.time() - oldest) + 0.1
                else:
                    wait_time = 0.1
            if wait_time > 0:
                time.sleep(min(wait_time, 1.0))
        raise RuntimeError("RateLimiter has been shut down")
    def make_request(self, url, max_retries=3, timeout=5):
        """发送 API 请求，带重试机制"""
        self._wait_for_slot()
        for attempt in range(max_retries):
            try:
                print(f"[尝试 {attempt + 1}/{max_retries}] 发送请求到: {url}")
                response = requests.get(url, timeout=timeout)
                response.raise_for_status()
                print(f"✓ 请求成功: {url} (状态码: {response.status_code})")
                return response
            except requests.exceptions.Timeout:
                print(f"⚠ 请求超时: {url}")
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
            except requests.exceptions.ConnectionError as e:
                print(f"⚠ 连接错误: {url} - {e}")
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
            except requests.exceptions.HTTPError as e:
                print(f"✗ HTTP 错误: {url} - {e}")
                raise
        return None
    def shutdown(self):
        """优雅关闭速率限制器"""
        self._shutdown = True
        print("RateLimiter 已关闭")
# 限制每秒最多 5 个请求
limiter = RateLimiter(max_requests=5, time_window=1)
def fetch_data(item_id):
    """获取数据的 worker 函数"""
    try:
        url = f"https://api.example.com/items/{item_id}"
        response = limiter.make_request(url)
        if response:
            print(f"✅ 获取到数据: {item_id}, 大小: {len(response.content)} bytes")
    except Exception as e:
        print(f"❌ 获取数据失败 {item_id}: {e}")
try:
    threads = [threading.Thread(target=fetch_data, args=(i,), daemon=True)
               for i in range(20)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)
    print("\n所有请求完成")
finally:
    limiter.shutdown()

1. 滑动窗口算法替代 Timer
# ❌ 原代码：创建大量 Timer 线程
threading.Timer(self.time_window, self.semaphore.release).start()
# ✅ 优化后：使用 deque 记录请求时间，自动清理过期记录
self.request_times = deque()
self._clean_old_requests()  # 移除超出时间窗口的记录
优势：
不会创建额外线程，避免资源泄漏
更精确的速率控制
支持动态调整时间窗口
2. 指数退避重试机制
for attempt in range(max_retries):
    try:
        response = requests.get(url, timeout=timeout)
        response.raise_for_status()
        return response
    except requests.exceptions.Timeout:
        if attempt == max_retries - 1:
            raise
        time.sleep(2 ** attempt)  # 1s, 2s, 4s...
3. 优雅关闭支持
def shutdown(self):
    self._shutdown = True
# 使用时
try:
    # ... 执行任务 ...
finally:
    limiter.shutdown()  # 确保资源清理
4. 守护线程 + 超时控制
threads = [threading.Thread(target=fetch_data, args=(i,), daemon=True) 
           for i in range(20)]
for t in threads:
    t.join(timeout=30)  # 最多等待 30 秒

import threading
import time
import requests
class TokenBucketRateLimiter:
    """基于令牌桶算法的速率限制器"""
        def __init__(self, rate=5, capacity=10):
        """
        :param rate: 每秒生成的令牌数
        :param capacity: 桶的最大容量
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.time()
        self.lock = threading.Lock()
        self._shutdown = False
        def _refill(self):
        """补充令牌"""
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now
        def acquire(self, timeout=None):
        """获取令牌"""
        deadline = time.time() + timeout if timeout else None
                while not self._shutdown:
            with self.lock:
                self._refill()
                if self.tokens >= 1:
                    self.tokens -= 1
                    return True
                        if deadline and time.time() > deadline:
                return False
                        time.sleep(0.1)
                raise RuntimeError("TokenBucketRateLimiter has been shut down")
        def make_request(self, url, timeout=5, max_retries=3):
        """发送请求"""
        if not self.acquire(timeout=10):
            raise TimeoutError("等待令牌超时")
                for attempt in range(max_retries):
            try:
                response = requests.get(url, timeout=timeout)
                response.raise_for_status()
                return response
            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(min(2 ** attempt, 10))
        def shutdown(self):
        self._shutdown = True
# 使用：每秒 5 个请求，峰值可达 10 个
limiter = TokenBucketRateLimiter(rate=5, capacity=10)
def fetch_data(item_id):
    try:
        url = f"https://api.example.com/items/{item_id}"
        response = limiter.make_request(url)
        print(f"✅ {item_id}: {len(response.content)} bytes")
    except Exception as e:
        print(f"❌ {item_id}: {e}")
threads = [threading.Thread(target=fetch_data, args=(i,), daemon=True) 
           for i in range(20)]
try:
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)
finally:
    limiter.shutdown()