Ba tháng trước, tôi nhận được cuộc gọi lúc 2 giờ sáng từ đội vận hành. Website thương mại điện tử của khách hàng — một nền tảng bán đồ điện tử với 50,000 người dùng đồng thời — bị sập hoàn toàn. Nguyên nhân? Hệ thống AI chatbot đang xử lý 2,000 concurrent requests nhưng không có cơ chế queue, dẫn đến timeout chuỗi và crash toàn bộ service.
Kể từ ngày đó, tôi đã xây dựng và tối ưu hóa hệ thống xử lý concurrent request cho hơn 20 dự án AI API. Trong bài viết này, tôi sẽ chia sẻ toàn bộ kiến thức thực chiến — từ những lỗi đau thương nhất đến giải pháp đã giúp tiết kiệm 85% chi phí với HolySheep AI.
Tại Sao Concurrent Request Quan Trọng?
Khi bạn xây dựng ứng dụng sử dụng AI API (chatbot, RAG system, auto-complete), người dùng không đợi nhau. Nếu 100 người click cùng lúc, hệ thống của bạn phải xử lý 100 requests đồng thời. Không có chiến lược đúng, bạn sẽ gặp:
- Rate Limiting: API provider chặn request khi vượt ngưỡng
- Timeout Chain: Request chờ quá lâu, client ngắt kết nối
- Memory Explosion: Tất cả requests giữ kết nối, RAM tăng không kiểm soát
- Chi phí phí oan: Retry thất bại = trả tiền cho request bị lãng phí
Cấu Trúc Cơ Bản: Semaphore + Retry
Đây là pattern đầu tiên tôi triển khai cho mọi dự án. Với HolySheep AI, tỷ giá chỉ ¥1=$1 và độ trễ trung bình dưới 50ms, bạn có thể xử lý khối lượng lớn mà không lo về chi phí.
import asyncio
import aiohttp
from aiohttp import ClientTimeout
from typing import List, Dict, Any
import time
class HolySheepAIClient:
"""
HolySheep AI Client - Xử lý concurrent request với semaphore pattern
Giá 2026: GPT-4.1 $8/MTok, DeepSeek V3.2 $0.42/MTok (tiết kiệm 85%+)
"""
def __init__(
self,
api_key: str,
max_concurrent: int = 10,
timeout_seconds: int = 30,
max_retries: int = 3
):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.semaphore = asyncio.Semaphore(max_concurrent)
self.timeout = ClientTimeout(total=timeout_seconds)
self.max_retries = max_retries
# Metrics cho monitoring
self.stats = {
"total_requests": 0,
"successful": 0,
"failed": 0,
"retried": 0
}
async def _make_request(
self,
session: aiohttp.ClientSession,
endpoint: str,
payload: Dict[str, Any]
) -> Dict[str, Any]:
"""Thực hiện single request với retry logic"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
for attempt in range(self.max_retries):
try:
async with self.semaphore: # Kiểm soát concurrency
async with session.post(
f"{self.base_url}{endpoint}",
json=payload,
headers=headers,
timeout=self.timeout
) as response:
self.stats["total_requests"] += 1
if response.status == 200:
self.stats["successful"] += 1
return await response.json()
elif response.status == 429: # Rate limit
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
continue
elif response.status >= 500: # Server error
await asyncio.sleep(1 * attempt)
continue
else:
error_data = await response.json()
raise Exception(f"API Error {response.status}: {error_data}")
except asyncio.TimeoutError:
if attempt == self.max_retries - 1:
raise Exception("Request timeout after retries")
await asyncio.sleep(1)
except Exception as e:
if attempt == self.max_retries - 1:
raise
self.stats["retried"] += 1
async def chat_completions(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
**kwargs
) -> Dict[str, Any]:
"""Gửi single chat completion request"""
payload = {
"model": model,
"messages": messages,
**kwargs
}
async with aiohttp.ClientSession() as session:
return await self._make_request(session, "/chat/completions", payload)
async def batch_chat(
self,
batch_requests: List[Dict[str, Any]],
model: str = "gpt-4.1"
) -> List[Dict[str, Any]]:
"""
Xử lý batch requests đồng thời với concurrency control
Đây là cách tôi xử lý 2,000 requests/giây cho hệ thống e-commerce
"""
tasks = []
for req in batch_requests:
task = self.chat_completions(
messages=req["messages"],
model=model,
temperature=req.get("temperature", 0.7)
)
tasks.append(task)
# Chạy tất cả requests đồng thời, semaphore kiểm soát số lượng
results = await asyncio.gather(*tasks, return_exceptions=True)
# Xử lý exceptions
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append({
"error": str(result),
"request_index": i,
"status": "failed"
})
else:
processed_results.append({
"data": result,
"request_index": i,
"status": "success"
})
return processed_results
def get_stats(self) -> Dict[str, int]:
"""Lấy thống kê request"""
return self.stats.copy()
=== SỬ DỤNG THỰC TẾ ===
async def main():
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=20, # Tối ưu cho HolySheep API
timeout_seconds=30,
max_retries=3
)
# Tạo 100 requests giả lập cho batch processing
batch = []
for i in range(100):
batch.append({
"messages": [
{"role": "system", "content": "Bạn là trợ lý bán hàng"},
{"role": "user", "content": f"Tư vấn sản phẩm #{i+1}"}
],
"temperature": 0.7
})
print(f"Processing {len(batch)} concurrent requests...")
start = time.time()
results = await client.batch_chat(batch, model="deepseek-v3.2") # $0.42/MTok
elapsed = time.time() - start
stats = client.get_stats()
print(f"\n=== KẾT QUẢ ===")
print(f"Thời gian xử lý: {elapsed:.2f}s")
print(f"Tổng requests: {stats['total_requests']}")
print(f"Thành công: {stats['successful']}")
print(f"Thất bại: {stats['failed']}")
print(f"Retry: {stats['retried']}")
if __name__ == "__main__":
asyncio.run(main())Worker Queue Pattern: Xử Lý 10,000+ Requests
Với dự án RAG enterprise của tôi (hệ thống tìm kiếm tài liệu cho công ty luật 500 nhân viên), batch size lên tới 10,000 documents. Tôi cần một hệ thống queue thực sự — không chỉ concurrency control mà còn job scheduling, priority queue, và dead letter handling.
import asyncio
import aiohttp
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Callable, Any, List
from enum import Enum
import time
import json
from datetime import datetime
class JobStatus(Enum):
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
RETRY = "retry"
@dataclass
class AIJob:
"""Job wrapper cho queue system"""
job_id: str
payload: dict
priority: int = 0 # 0 = thấp, 10 = cao
status: JobStatus = JobStatus.PENDING
retries: int = 0
max_retries: int = 3
result: Optional[Any] = None
error: Optional[str] = None
created_at: float = field(default_factory=time.time)
completed_at: Optional[float] = None
def __lt__(self, other):
# Priority queue: job có priority cao hơn được ưu tiên
return self.priority > other.priority
class AsyncWorkerQueue:
"""
Async Worker Queue cho AI API requests
- Priority queue support
- Auto-retry với exponential backoff
- Batch processing
- Progress tracking
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_workers: int = 50,
max_queue_size: int = 100000,
rate_limit_per_second: int = 100
):
self.api_key = api_key
self.base_url = base_url
# Queue system
self.jobs: deque[AIJob] = deque()
self.completed_jobs: List[AIJob] = []
self.failed_jobs: List[AIJob] = []
# Worker configuration
self.max_workers = max_workers
self.max_queue_size = max_queue_size
self.rate_limit = asyncio.Semaphore(rate_limit_per_second)
# Control flags
self._running = False
self._shutdown = False
# Metrics
self.metrics = {
"jobs_enqueued": 0,
"jobs_completed": 0,
"jobs_failed": 0,
"total_tokens": 0,
"avg_latency_ms": 0
}
# Callbacks
self.on_complete: Optional[Callable] = None
self.on_error: Optional[Callable] = None
def enqueue(
self,
payload: dict,
priority: int = 5,
job_id: Optional[str] = None
) -> str:
"""Thêm job vào queue"""
if len(self.jobs) >= self.max_queue_size:
raise Exception(f"Queue full! Max size: {self.max_queue_size}")
job_id = job_id or f"job_{int(time.time() * 1000)}"
job = AIJob(
job_id=job_id,
payload=payload,
priority=priority
)
# Insert theo priority (sort)
inserted = False
for i, existing_job in enumerate(self.jobs):
if job.priority > existing_job.priority:
self.jobs.insert(i, job)
inserted = True
break
if not inserted:
self.jobs.append(job)
self.metrics["jobs_enqueued"] += 1
return job_id
async def _process_single_job(
self,
job: AIJob,
session: aiohttp.ClientSession
) -> AIJob:
"""Xử lý một job đơn lẻ"""
job.status = JobStatus.PROCESSING
start_time = time.time()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
endpoint = job.payload.get("endpoint", "/chat/completions")
data = job.payload.get("data", job.payload)
try:
async with self.rate_limit: # Rate limiting
async with session.post(
f"{self.base_url}{endpoint}",
json=data,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 200:
result = await response.json()
job.result = result
job.status = JobStatus.COMPLETED
job.completed_at = time.time()
# Update metrics
if "usage" in result:
tokens = result["usage"].get("total_tokens", 0)
self.metrics["total_tokens"] += tokens
latency = (time.time() - start_time) * 1000
self.metrics["avg_latency_ms"] = (
(self.metrics["avg_latency_ms"] * self.metrics["jobs_completed"] + latency) /
(self.metrics["jobs_completed"] + 1)
)
self.metrics["jobs_completed"] += 1
if self.on_complete:
await self.on_complete(job)
elif response.status == 429:
# Rate limit - retry với backoff
job.retries += 1
if job.retries < job.max_retries:
job.status = JobStatus.RETRY
wait_time = 2 ** job.retries
await asyncio.sleep(wait_time)
self.jobs.append(job) # Re-queue
else:
job.status = JobStatus.FAILED
job.error = "Rate limit exceeded after retries"
self.metrics["jobs_failed"] += 1
if self.on_error:
await self.on_error(job)
else:
job.status = JobStatus.FAILED
error_data = await response.text()
job.error = f"HTTP {response.status}: {error_data}"
self.metrics["jobs_failed"] += 1
if self.on_error:
await self.on_error(job)
except asyncio.TimeoutError:
job.retries += 1
if job.retries < job.max_retries:
job.status = JobStatus.RETRY
await asyncio.sleep(2 ** job.retries)
self.jobs.append(job)
else:
job.status = JobStatus.FAILED
job.error = "Timeout after retries"
self.metrics["jobs_failed"] += 1
except Exception as e:
job.status = JobStatus.FAILED
job.error = str(e)
self.metrics["jobs_failed"] += 1
if self.on_error:
await self.on_error(job)
return job
async def _worker(
self,
worker_id: int,
session: aiohttp.ClientSession
):
"""Worker process chạy liên tục"""
while not self._shutdown:
job = None
# Lấy job từ queue
if self.jobs:
job = self.jobs.popleft()
if job:
await self._process_single_job(job, session)
else:
# Không có job, đợi một chút
await asyncio.sleep(0.1)
async def start(self):
"""Khởi động worker pool"""
self._running = True
connector = aiohttp.TCPConnector(limit=100)
async with aiohttp.ClientSession(connector=connector) as session:
# Tạo workers
workers = [
asyncio.create_task(self._worker(i, session))
for i in range(self.max_workers)
]
print(f"Worker Queue started with {self.max_workers} workers")
# Monitor cho đến khi shutdown
while self._running and not self._shutdown:
await asyncio.sleep(5)
queue_size = len(self.jobs)
if queue_size % 100 == 0 and queue_size > 0:
print(f"Queue size: {queue_size}, Completed: {self.metrics['jobs_completed']}")
# Cancel workers
for w in workers:
w.cancel()
def stop(self):
"""Dừng worker pool"""
self._running = False
self._shutdown = True
def get_stats(self) -> dict:
"""Lấy thống kê hệ thống"""
return {
"queue_size": len(self.jobs),
"completed": self.metrics["jobs_completed"],
"failed": self.metrics["jobs_failed"],
"total_tokens": self.metrics["total_tokens"],
"avg_latency_ms": round(self.metrics["avg_latency_ms"], 2),
"estimated_cost_usd": round(self.metrics["total_tokens"] / 1_000_000 * 0.42, 2) # DeepSeek pricing
}
=== DEMO: Xử lý 10,000 documents cho RAG system ===
async def demo_rag_ingestion():
queue = AsyncWorkerQueue(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_workers=50,
rate_limit_per_second=100
)
# Enqueue 100 documents (demo - thực tế có thể là 10,000+)
documents = [
{"content": f"Hợp đồng mua bán số {i}", "metadata": {"type": "contract"}}
for i in range(100)
]
job_ids = []
for i, doc in enumerate(documents):
payload = {
"endpoint": "/embeddings",
"data": {
"model": "text-embedding-3-small",
"input": doc["content"]
}
}
job_id = queue.enqueue(payload, priority=5)
job_ids.append(job_id)
print(f"Enqueued {len(job_ids)} embedding jobs")
# Chạy queue trong 60 giây
try:
await queue.start()
except KeyboardInterrupt:
queue.stop()
stats = queue.get_stats()
print(f"\n=== RAG INGESTION RESULTS ===")
print(f"Documents processed: {stats['completed']}")
print(f"Failed: {stats['failed']}")
print(f"Total tokens: {stats['total_tokens']:,}")
print(f"Avg latency: {stats['avg_latency_ms']}ms")
print(f"Estimated cost (DeepSeek V3.2): ${stats['estimated_cost_usd']}")
if __name__ == "__main__":
asyncio.run(demo_rag_ingestion())Advanced: Connection Pooling và Circuit Breaker
Trong production, tôi đã gặp trường hợp HolySheep API tạm thời chậm do maintenance (thường chỉ 5-10 phút). Thay vì để tất cả requests fail, tôi implement circuit breaker pattern — tự động ngắt kết nối khi API có vấn đề và gradual recovery.
import asyncio
import aiohttp
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import random
class CircuitState(Enum):
CLOSED = "closed" # Bình thường, request đi qua
OPEN = "open" # Circuit break, request bị reject
HALF_OPEN = "half_open" # Thử recovery
@dataclass
class CircuitBreaker:
"""Circuit Breaker cho API calls"""
failure_threshold: int = 5 # Số lần fail để open circuit
success_threshold: int = 3 # Số lần success để close circuit
timeout_seconds: int = 30 # Thời gian open trước khi thử half-open
half_open_max_calls: int = 3 # Số calls trong half-open state
def __post_init__(self):
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
def record_success(self):
"""Ghi nhận thành công"""
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
self.half_open_calls += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.half_open_calls = 0
else:
self.failure_count = 0
def record_failure(self):
"""Ghi nhận thất bại"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.half_open_calls = 0
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def can_attempt(self) -> bool:
"""Kiểm tra xem có thể thử request không"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Kiểm tra timeout
if time.time() - self.last_failure_time >= self.timeout_seconds:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
self.success_count = 0
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.half_open_max_calls
return False
def get_state(self) -> str:
return self.state.value
class ResilientAIClient:
"""
AI Client với Connection Pooling + Circuit Breaker
Tối ưu cho production với HolySheep AI
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.base_url = base_url
# Connection pool
self.connector = aiohttp.TCPConnector(
limit=100, # Tổng connections
limit_per_host=50, # Connections per host
ttl_dns_cache=300, # DNS cache 5 phút
keepalive_timeout=30
)
# Circuit breaker
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
success_threshold=3,
timeout_seconds=30
)
# Retry configuration
self.max_retries = 3
self.retry_delays = [1, 2, 4] # Exponential backoff
# Metrics
self.request_count = 0
self.cache_hits = 0
self.cache: dict = {} # Simple in-memory cache
def _get_cache_key(self, payload: dict) -> str:
"""Tạo cache key từ payload"""
import hashlib
content = json.dumps(payload, sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()
async def chat(
self,
messages: list,
model: str = "deepseek-v3.2", # $0.42/MTok - best value
use_cache: bool = True,
**kwargs
) -> dict:
"""
Gửi chat request với resilience features
"""
payload = {
"model": model,
"messages": messages,
**kwargs
}
# Check cache
if use_cache:
cache_key = self._get_cache_key(payload)
if cache_key in self.cache:
self.cache_hits += 1
return self.cache[cache_key]
# Check circuit breaker
if not self.circuit_breaker.can_attempt():
raise Exception(f"Circuit breaker OPEN. State: {self.circuit_breaker.get_state()}")
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with aiohttp.ClientSession(connector=self.connector) as session:
for attempt in range(self.max_retries):
try:
self.request_count += 1
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 200:
result = await response.json()
self.circuit_breaker.record_success()
# Store in cache
if use_cache:
self.cache[cache_key] = result
return result
elif response.status == 429:
# Rate limit - retry sau
delay = self.retry_delays[min(attempt, len(self.retry_delays)-1)]
await asyncio.sleep(delay)
continue
elif response.status >= 500:
# Server error - retry
delay = self.retry_delays[min(attempt, len(self.retry_delays)-1)]
await asyncio.sleep(delay)
continue
else:
error = await response.text()
self.circuit_breaker.record_failure()
raise Exception(f"API error {response.status}: {error}")
except asyncio.TimeoutError:
if attempt == self.max_retries - 1:
self.circuit_breaker.record_failure()
raise Exception("Request timeout")
await asyncio.sleep(self.retry_delays[attempt])
except aiohttp.ClientError as e:
self.circuit_breaker.record_failure()
if attempt == self.max_retries - 1:
raise
await asyncio.sleep(self.retry_delays[attempt])
self.circuit_breaker.record_failure()
raise Exception("Max retries exceeded")
def get_circuit_state(self) -> str:
return self.circuit_breaker.get_state()
def get_metrics(self) -> dict:
return {
"total_requests": self.request_count,
"cache_hits": self.cache_hits,
"cache_hit_rate": f"{self.cache_hits/max(self.request_count, 1)*100:.1f}%",
"circuit_state": self.get_circuit_state(),
"circuit_failures": self.circuit_breaker.failure_count
}
=== TEST CIRCUIT BREAKER ===
async def test_circuit_breaker():
client = ResilientAIClient("YOUR_HOLYSHEEP_API_KEY")
# Simulate failures để trigger circuit breaker
async def simulate_requests():
for i in range(20):
try:
# Giả lập request (thực tế gọi real API)
if i < 5: # 5 requests đầu fail
raise Exception("Simulated failure")
# Success sau đó
result = await client.chat([
{"role": "user", "content": f"Request {i}"}
])
print(f"✓ Request {i} successful")
except Exception as e:
print(f"✗ Request {i} failed: {e}")
print(f" Circuit state: {client.get_circuit_state()}")
await asyncio.sleep(0.5)
await simulate_requests()
print("\n=== METRICS ===")
print(client.get_metrics())
if __name__ == "__main__":
import json
asyncio.run(test_circuit_breaker())So Sánh Chi Phí: HolySheep vs Providers Khác
Một trong những lý do tôi chọn HolySheep AI cho các dự án production là tỷ giá ¥1=$1 — tiết kiệm 85%+ so với OpenAI hay Anthropic. Bảng so sánh chi phí xử lý 1 triệu tokens:
| Provider | Model | Giá/MTok | Chi phí cho 1M tokens | Concurrent capacity |
|---|---|---|---|---|
| OpenAI | GPT-4.1 | $8.00 | $8.00 | 500 rpm |
| Anthropic | Claude Sonnet 4.5 | $15.00 | $15.00 | 200 rpm |
| Gemini 2.5 Flash | $2.50 | $2.50 | 1000 rpm | |
| HolySheep | DeepSeek V3.2 | $0.42 | $0.42 | 2000 rpm |
Với hệ thống e-commerce xử lý 10 triệu tokens/tháng, bạn tiết kiệm $75,800 khi dùng HolySheep thay vì Claude Sonnet!
Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "Connection pool exhausted"
Mô tả: Khi số lượng concurrent requests tăng đột biến, aiohttp báo lỗi "Cannot connect to host, connection pool limit reached".
# ❌ SAI: Không giới hạn connection pool
async with aiohttp.ClientSession() as session:
async with session.post(url, json=data) as response:
...
✅ ĐÚNG: Cấu hình connection pool phù hợp
connector = aiohttp.TCPConnector(
limit=100, # Tổng số connections trong pool
limit_per_host=50, # Connections giới hạn per host
ttl_dns_cache=300 # Cache DNS để giảm overhead
)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.post(url, json=data) as response:
...
✅ TỐI ƯU: Thêm keepalive để reuse connections
connector = aiohttp.TCPConnector(
limit=200,
limit_per_host=100,
keepalive_timeout=30
)2. Lỗi "Rate limit exceeded" Xử Lý Không Đúng
Mô tả: Request bị chặn với HTTP 429, nhưng retry không hoạt động đúng cách.
# ❌ SAI: Retry ngay lập tức không có backoff
for attempt in range(3):
try:
response = await session.post(url, json=data)
if response.status == 429:
continue # Retry ngay = vẫn bị 429
except Exception as e:
print(e)
✅ ĐÚNG: Exponential backoff với jitter
import random
async def retry_with_backoff(session, url, data, max_retries=5):
for attempt in range(max_retries):
try:
response = await session.post(url, json=data)
if response.status == 200:
return await response.json()
elif response.status == 429:
# Lấy retry-after từ header nếu có
retry_after = response.headers.get('Retry-After')
if retry_after:
wait_time = int(retry_after)
else:
# Exponential backoff + random jitter
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"Rate limited. Waiting {wait_time:.2f}s...")
await asyncio.sleep(wait_time)
else:
response.raise_for_status()
except Exception as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
raise Exception("Max retries exceeded")3. Lỗi Memory Leak Khi Xử Lý Batch Lớn
Mô tả: Xử lý 10,000+ requests cùng lúc làm RAM tăng liên tục cho đến khi OOM.
# ❌ SAI: Load tất cả results vào memory
all_results = []
for item in huge_batch:
result = await process_item(item)
all_results.append(result) # Memory grows unbounded
✅ ĐÚNG: Stream processing với chunking
CHUNK_SIZE = 100
async def process_in_chunks(items, process_fn):
"""Xử lý batch lớn theo chunks để tránh memory leak"""
all_results = []