# 模拟并发流管理器
import threading
from collections import deque

class ConcurrencyManager:
    def __init__(self, max_concurrent: int = 100):
        self.max_concurrent = max_concurrent
        self.active_streams: set = set()
        self.waiting_queue: deque = deque()
        self.lock = threading.Lock()
        self.condition = threading.Condition(self.lock)
    
    def acquire(self, stream_id: int) -> bool:
        """获取流的并发许可"""
        with self.condition:
            if len(self.active_streams) < self.max_concurrent:
                self.active_streams.add(stream_id)
                return True
            else:
                self.waiting_queue.append(stream_id)
                return False
    
    def release(self, stream_id: int):
        """释放流的并发占用"""
        with self.condition:
            self.active_streams.discard(stream_id)
            self.condition.notify()
    
    def wait_for_slot(self, stream_id: int):
        """等待可用的并发槽位"""
        with self.condition:
            while len(self.active_streams) >= self.max_concurrent:
                self.condition.wait()
            self.active_streams.add(stream_id)

# 使用示例
manager = ConcurrencyManager(max_concurrent=100)
manager.acquire(1)   # stream 1 获得许可
manager.acquire(3)   # stream 3 获得许可
manager.release(1)   # stream 1 释放

package main

import (
	"fmt"
	"sync"
)

// Stream 表示一个 HTTP/2 流
type Stream struct {
	ID           uint32
	State        string
	HeadersSent  bool
	DataComplete bool
	mu           sync.Mutex
}

// Connection 管理多个流
type Connection struct {
	streams      map[uint32]*Stream
	maxStreams   int
	mu           sync.RWMutex
}

func NewConnection(maxStreams int) *Connection {
	return &Connection{
		streams:    make(map[uint32]*Stream),
		maxStreams: maxStreams,
	}
}

func (c *Connection) CreateStream(id uint32) (*Stream, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	
	if len(c.streams) >= c.maxStreams {
		return nil, fmt.Errorf("超出最大并发流限制 (%d)", c.maxStreams)
	}
	
	if _, exists := c.streams[id]; exists {
		return nil, fmt.Errorf("流 %d 已存在", id)
	}
	
	stream := &Stream{ID: id, State: "open"}
	c.streams[id] = stream
	return stream, nil
}

func (c *Connection) CloseStream(id uint32) {
	c.mu.Lock()
	defer c.mu.Unlock()
	delete(c.streams, id)
}

func (c *Connection) ActiveStreams() int {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return len(c.streams)
}

func main() {
	conn := NewConnection(100)
	
	// 模拟并发请求
	for i := uint32(1); i <= 10; i += 2 {
		stream, err := conn.CreateStream(i)
		if err != nil {
			fmt.Printf("创建流失败: %v\n", err)
			continue
		}
		fmt.Printf("创建流 %d，当前活跃: %d\n", stream.ID, conn.ActiveStreams())
	}
}

# 模拟对比实验
import time
import concurrent.futures
from dataclasses import dataclass
from typing import List

@dataclass
class Request:
    id: int
    path: str
    delay_ms: int  # 服务器处理耗时

def simulate_http11(requests: List[Request], max_conn: int = 6) -> dict:
    """模拟 HTTP/1.1 请求（受限于连接数和队头阻塞）"""
    results = {}
    start = time.time()
    
    # 分批处理（每批最多 max_conn 个）
    for i in range(0, len(requests), max_conn):
        batch = requests[i:i+max_conn]
        # 每个连接串行处理
        for req in batch:
            time.sleep(req.delay_ms / 1000)
            results[req.id] = time.time() - start
    
    return results

def simulate_http2(requests: List[Request]) -> dict:
    """模拟 HTTP/2 请求（真正的多路复用）"""
    results = {}
    start = time.time()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
        futures = {}
        for req in requests:
            future = executor.submit(lambda r: time.sleep(r.delay_ms/1000), req)
            futures[future] = req
        
        for future in concurrent.futures.as_completed(futures):
            req = futures[future]
            results[req.id] = time.time() - start
    
    return results

# 测试场景：10 个请求，处理时间各异
requests = [
    Request(1, "/css/style.css", 50),
    Request(2, "/js/app.js", 100),
    Request(3, "/api/users", 200),
    Request(4, "/api/orders", 300),  # 最慢
    Request(5, "/image/logo.png", 30),
    Request(6, "/js/utils.js", 60),
    Request(7, "/api/products", 150),
    Request(8, "/css/theme.css", 40),
    Request(9, "/image/bg.jpg", 80),
    Request(10, "/api/config", 20),
]

print("=== HTTP/1.1 (6 并发连接) ===")
h1_results = simulate_http11(requests, max_conn=6)
for rid, t in sorted(h1_results.items()):
    print(f"  请求 {rid}: {t:.3f}s")

print("\n=== HTTP/2 (多路复用) ===")
h2_results = simulate_http2(requests)
for rid, t in sorted(h2_results.items()):
    print(f"  请求 {rid}: {t:.3f}s")

package main

import (
	"fmt"
	"net/http"
)

func main() {
	// Go 的 HTTP/2 支持通过 ResponseWriter 设置优先级
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		// 获取流的优先级信息（来自客户端的 PRIORITY 帧）
		priority := r.Header.Get("Priority")
		fmt.Printf("流优先级: %s\n", priority)
		
		// 不同资源类型返回不同的缓存策略
		switch r.URL.Path {
		case "/critical.css":
			// CSS 关键资源，高优先级
			w.Header().Set("Content-Type", "text/css")
			w.Header().Set("Cache-Control", "public, max-age=31536000")
			w.Write([]byte("body { margin: 0; }"))
			
		case "/analytics.js":
			// 分析脚本，低优先级
			w.Header().Set("Content-Type", "application/javascript")
			w.Header().Set("Cache-Control", "public, max-age=3600")
			w.Write([]byte("// analytics code"))
			
		default:
			w.Write([]byte("Hello"))
		}
	})
}

// 服务端并发流配置
package main

import (
	"log"
	"net/http"
)

func main() {
	server := &http.Server{
		Addr: ":8443",
		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.Write([]byte("Hello"))
		}),
		// 最大并发流数（通过 SETTINGS 帧告知客户端）
		MaxConcurrentStreams: 100,
	}
	
	log.Printf("启动 HTTP/2 服务器，最大并发流: %d", 100)
	log.Fatal(server.ListenAndServeTLS("cert.pem", "key.pem"))
}

# 根据流优先级分配服务器资源
import asyncio
from enum import IntEnum
from dataclasses import dataclass

class Priority(IntEnum):
    CRITICAL = 0    # 关键渲染路径
    HIGH = 1        # 重要资源
    MEDIUM = 2      # 普通资源
    LOW = 3         # 非关键资源
    BACKGROUND = 4  # 后台任务

@dataclass
class StreamRequest:
    stream_id: int
    path: str
    priority: Priority

class StreamScheduler:
    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.queues = {p: asyncio.Queue() for p in Priority}
        self.workers = []
    
    async def submit(self, request: StreamRequest):
        await self.queues[request.priority].put(request)
    
    async def process(self):
        """按优先级顺序处理请求"""
        while True:
            # 优先处理高优先级队列
            for priority in Priority:
                if not self.queues[priority].empty():
                    request = await self.queues[priority].get()
                    await self.handle_request(request)
                    return
            await asyncio.sleep(0.001)
    
    async def handle_request(self, request: StreamRequest):
        # 模拟处理
        await asyncio.sleep(0.01)
        print(f"处理流 {request.stream_id}: {request.path} (优先级: {request.priority.name})")

# 使用示例
async def main():
    scheduler = StreamScheduler()
    
    requests = [
        StreamRequest(1, "/css/critical.css", Priority.CRITICAL),
        StreamRequest(3, "/api/users", Priority.HIGH),
        StreamRequest(5, "/image/logo.png", Priority.MEDIUM),
        StreamRequest(7, "/analytics.js", Priority.LOW),
    ]
    
    for req in requests:
        await scheduler.submit(req)
    
    for _ in requests:
        await scheduler.process()

asyncio.run(main())

// 微服务间 HTTP/2 通信示例
package main

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"
)

type ServiceClient struct {
	client *http.Client
	baseURL string
}

func NewServiceClient(baseURL string) *ServiceClient {
	transport := &http.Transport{
		MaxIdleConns:        100,
		MaxIdleConnsPerHost: 100,
		IdleConnTimeout:     90 * time.Second,
	}
	
	return &ServiceClient{
		client:  &http.Client{Transport: transport, Timeout: 10 * time.Second},
		baseURL: baseURL,
	}
}

func (c *ServiceClient) Call(ctx context.Context, path string) ([]byte, error) {
	req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+path, nil)
	if err != nil {
		return nil, err
	}
	
	resp, err := c.client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	
	return io.ReadAll(resp.Body)
}

// 并行调用多个微服务
func CallServices(ctx context.Context, client *ServiceClient, paths []string) (map[string][]byte, error) {
	results := make(map[string][]byte)
	var mu sync.Mutex
	var wg sync.WaitGroup
	errs := make(chan error, len(paths))
	
	for _, path := range paths {
		wg.Add(1)
		go func(p string) {
			defer wg.Done()
			data, err := client.Call(ctx, p)
			if err != nil {
				errs <- err
				return
			}
			mu.Lock()
			results[p] = data
			mu.Unlock()
		}(path)
	}
	
	wg.Wait()
	close(errs)
	
	for err := range errs {
		return nil, err
	}
	return results, nil
}

func main() {
	client := NewServiceClient("https://api.example.com")
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	
	services := []string{
		"/users/123",
		"/orders/recent",
		"/products/search?q=shoes",
	}
	
	start := time.Now()
	results, err := CallServices(ctx, client, services)
	if err != nil {
		fmt.Printf("调用失败: %v\n", err)
		return
	}
	
	fmt.Printf("调用完成，耗时: %v\n", time.Since(start))
	for path, data := range results {
		fmt.Printf("  %s: %d bytes\n", path, len(data))
	}
}