// Go Channel 天然支持背压
// 当缓冲区满时，发送方阻塞

func producer(ch chan<- int) {
    for i := 0; ; i++ {
        ch <- i  // 如果缓冲区满，这里会阻塞 → 天然背压
        fmt.Printf("生产: %d\n", i)
    }
}

func consumer(ch <-chan int) {
    for v := range ch {
        time.Sleep(100 * time.Millisecond)  // 慢消费者
        fmt.Printf("消费: %d\n", v)
    }
}

func main() {
    ch := make(chan int, 10)  // 缓冲区大小 10
    go producer(ch)
    consumer(ch)
}

import asyncio

async def bounded_pipeline(source, max_in_flight: int = 10):
    semaphore = asyncio.Semaphore(max_in_flight)
    
    async def process_item(item):
        async with semaphore:  # 限制并发数
            result = await expensive_operation(item)
            return result
    
    tasks = [process_item(item) async for item in source]
    return await asyncio.gather(*tasks)

const { Readable, Transform, Writable } = require('stream');

// Node.js Stream 内置背压
const readable = new Readable({
    read(size) {
        // 当消费者消费完后，这里会被调用
        // 生产更多数据
    }
});

const writable = new Writable({
    write(chunk, encoding, callback) {
        // 如果写入速度跟不上
        // writable.write() 返回 false
        // readable 自动暂停
        slowDatabaseInsert(chunk, callback);
    }
});

readable.pipe(writable);
// Stream.pipe() 自动处理背压

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // 带缓冲的通道，支持背压
    let (tx, mut rx) = mpsc::channel(100);

    // 生产者
    tokio::spawn(async move {
        for i in 0..10000 {
            tx.send(i).await.unwrap();  // 缓冲区满时阻塞
        }
    });

    // 慢消费者
    while let Some(value) = rx.recv().await {
        tokio::time::sleep(Duration::from_millis(100)).await;
        println!("处理: {}", value);
    }
}

// 丢弃新数据（Drop Newest）
func dropNewestProducer(ch chan<- int, value int) bool {
    select {
    case ch <- value:
        return true  // 发送成功
    default:
        return false  // 缓冲区满，丢弃
    }
}

// 丢弃旧数据（Drop Oldest）
func dropOldestProducer(ch chan int, value int) {
    select {
    case ch <- value:
        return  // 发送成功
    default:
        select {
        case <-ch:  // 丢弃一个旧数据
        default:
        }
        ch <- value  // 再尝试发送
    }
}

class AdaptiveRateLimiter:
    """自适应速率限制器"""
    
    def __init__(self, initial_rate: float = 100.0):
        self.rate = initial_rate
        self.min_rate = 10.0
        self.max_rate = 10000.0
        self.success_count = 0
        self.error_count = 0
        self.window_size = 100
    
    def on_success(self):
        self.success_count += 1
        self._adjust()
    
    def on_error(self):
        self.error_count += 1
        self._adjust()
    
    def _adjust(self):
        total = self.success_count + self.error_count
        if total >= self.window_size:
            error_rate = self.error_count / total
            if error_rate > 0.1:  # 错误率 > 10%，降低速率
                self.rate = max(self.min_rate, self.rate * 0.8)
            elif error_rate < 0.01:  # 错误率 < 1%，提高速率
                self.rate = min(self.max_rate, self.rate * 1.1)
            self.success_count = 0
            self.error_count = 0
    
    async def acquire(self):
        interval = 1.0 / self.rate
        await asyncio.sleep(interval)

import asyncio
from collections import deque

class LogCollector:
    def __init__(self, max_buffer: int = 10000, flush_interval: float = 1.0):
        self.buffer = deque(maxlen=max_buffer)  # 有界缓冲，丢弃旧数据
        self.max_buffer = max_buffer
        self.flush_interval = flush_interval
        self.lock = asyncio.Lock()
    
    async def ingest(self, log_entry: dict):
        """接收日志（可能非常快）"""
        async with self.lock:
            if len(self.buffer) >= self.max_buffer:
                dropped = self.buffer.popleft()  # 丢弃最旧的
                print(f"缓冲区满，丢弃旧日志: {dropped.get('id')}")
            self.buffer.append(log_entry)
    
    async def flush_loop(self):
        """定期刷写到存储"""
        while True:
            await asyncio.sleep(self.flush_interval)
            async with self.lock:
                if not self.buffer:
                    continue
                batch = list(self.buffer)
                self.buffer.clear()
            
            try:
                await self.write_to_storage(batch)
            except Exception as e:
                print(f"刷写失败: {e}")
                # 可以重新放回缓冲区或写入死信队列
    
    async def write_to_storage(self, batch: list[dict]):
        """批量写入存储（有速率限制）"""
        await asyncio.sleep(0.1)  # 模拟写入延迟
        print(f"写入 {len(batch)} 条日志")