Khi tôi lần đầu xây dựng hệ thống trading bot cho thị trường crypto, đối mặt với hàng triệu tick data mỗi giây từ nhiều sàn giao dịch, tôi nhận ra rằng một REST API đơn giản sẽ không bao giờ đủ. Sau 3 tháng thử nghiệm với Kafka, WebSocket, và các giải pháp relay khác nhau, tôi đã tìm ra công thức tối ưu để xử lý real-time market data mà không bị bottleneck. Trong bài viết này, tôi sẽ chia sẻ toàn bộ kiến thức thực chiến, kèm theo so sánh chi phí và hiệu suất giữa các phương án để bạn có thể đưa ra quyết định đúng đắn cho dự án của mình.

Thực Trạng Xử Lý Tick Data Trong Thị Trường Crypto

Tick data (dữ liệu giao dịch tức thời) bao gồm price, volume, bid/ask của mỗi giao dịch. Với các sàn lớn như Binance, huobi, OKX, khối lượng data có thể lên tới 100,000+ messages/giây. Nếu bạn đang xây dựng trading bot, arbitrage system, hoặc data analytics platform, việc xử lý hiệu quả nguồn dữ liệu khổng lồ này là yếu tố sống còn. Dưới đây là bảng so sánh chi tiết giữa các phương án phổ biến nhất hiện nay:

Tiêu chí HolySheep AI Relay API Chính Thức (Binance/Kraken) WebSocket Direct Kafka + Custom Consumer
Độ trễ trung bình <50ms 100-300ms 20-50ms 5-15ms (local)
Setup thời gian 5 phút 30 phút 2-4 giờ 1-3 ngày
Độ phức tạp code Rất thấp Thấp Cao Rất cao
Quản lý connection Tự động Thủ công Thủ công + reconnect logic Quản lý cluster phức tạp
Chi phí hàng tháng Tín dụng miễn phí khi đăng ký Miễn phí (có rate limit) Miễn phí $50-500 (server + maintenance)
Reliability (SLA) 99.9% 99.5% Phụ thuộc infrastructure Tự quản lý
Hỗ trợ multi-exchange Có Không (cần SDK riêng) Không Cần custom integration

Như bạn thấy, mỗi giải pháp có trade-off riêng. Với những người mới bắt đầu hoặc dự án có budget hạn chế, HolySheep AI là lựa chọn tối ưu về độ trễ thấp và chi phí vận hành gần như bằng không. Tuy nhiên, nếu bạn cần kiểm soát hoàn toàn infrastructure hoặc xử lý hàng triệu messages/giây với latency cực thấp, Kafka vẫn là giải pháp mạnh nhất.

Tại Sao Cần Message Queue Cho Tick Data?

Trước khi đi vào chi tiết kỹ thuật, hãy hiểu tại sao message queue (hàng đợi tin nhắn) lại quan trọng trong xử lý tick data:

Kiến Trúc Hệ Thống Xử Lý Tick Data Với Kafka

Đây là kiến trúc mà tôi đã implement thành công cho nhiều dự án trading:

┌─────────────┐    WebSocket    ┌─────────────┐    Produce    ┌─────────────┐
│   Exchange  │ ──────────────► │   Kafka     │ ─────────────► │   Consumer  │
│  (Binance,  │                 │   Broker    │                │   Cluster   │
│   huobi)    │                 │  (Cluster)  │                │             │
└─────────────┘                 └─────────────┘                └─────────────┘
                                                                    │
                        ┌───────────────────────────────────────────┤
                        │                    │                      │
                        ▼                    ▼                      ▼
               ┌─────────────┐      ┌─────────────┐       ┌─────────────┐
               │  Trading    │      │  Real-time  │       │  Historical │
               │    Bot      │      │  Dashboard  │       │  Analytics  │
               └─────────────┘      └─────────────┘       └─────────────┘

Cài Đặt Môi Trường Kafka

Để bắt đầu, bạn cần cài đặt Kafka. Tôi khuyên dùng Docker Compose cho local development vì đơn giản và nhanh:

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
    networks:
      - kafka-network

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9091"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT,PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
    networks:
      - kafka-network

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    depends_on:
      - kafka
    networks:
      - kafka-network

networks:
  kafka-network:
    driver: bridge

Chạy lệnh sau để khởi động Kafka cluster:

docker-compose up -d


Kiểm tra Kafka đã sẵn sàng chưa

docker-compose logs kafka | grep "started (kafka.server.KafkaServer)"


Tạo topic cho tick data

docker exec kafka kafka-topics --create \
  --topic crypto-tick-btcusdt \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 1


Tạo topic cho aggregated data

docker exec kafka kafka-topics --create \
  --topic crypto-ohlcv-1m \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 1


Verify topics đã được tạo

docker exec kafka kafka-topics --list --bootstrap-server localhost:9092

Producer: Kết Nối Exchange Và Gửi Tick Data Vào Kafka

Tiếp theo, tôi sẽ implement producer để kết nối với exchange và gửi tick data vào Kafka. Dưới đây là production-ready code với error handling và reconnection logic:

# requirements: pip install kafka-python asyncio aiohttp

import asyncio
import json
import logging
from datetime import datetime
from kafka import KafkaProducer
from kafka.errors import KafkaError
import aiohttp

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CryptoTickProducer:
    def __init__(self, bootstrap_servers='localhost:9092'):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self.running = False
        self.exchanges = {
            'binance': {
                'ws_url': 'wss://stream.binance.com:9443/ws',
                'symbols': ['btcusdt', 'ethusdt', 'bnbusdt']
            },
            'huobi': {
                'ws_url': 'wss://api.huobi.pro/ws',
                'symbols': ['btcusdt', 'ethusdt']
            }
        }
        
    async def initialize(self):
        """Khởi tạo Kafka producer với retry logic"""
        max_retries = 5
        retry_delay = 2
        
        for attempt in range(max_retries):
            try:
                self.producer = KafkaProducer(
                    bootstrap_servers=self.bootstrap_servers,
                    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                    key_serializer=lambda k: k.encode('utf-8') if k else None,
                    acks='all',
                    retries=3,
                    max_in_flight_requests_per_connection=1,
                    compression_type='gzip'
                )
                logger.info("Kafka producer initialized successfully")
                return True
            except KafkaError as e:
                logger.warning(f"Attempt {attempt + 1}/{max_retries} failed: {e}")
                if attempt < max_retries - 1:
                    await asyncio.sleep(retry_delay * (attempt + 1))
                else:
                    logger.error("Failed to initialize Kafka producer after all retries")
                    return False
                    
    async def connect_binance(self, symbol):
        """Kết nối WebSocket với Binance"""
        uri = f"wss://stream.binance.com:9443/ws/{symbol}@trade"
        
        async with aiohttp.ClientSession() as session:
            async with session.ws_connect(uri) as ws:
                logger.info(f"Connected to Binance {symbol} stream")
                
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        data = json.loads(msg.data)
                        tick_data = self._normalize_binance_tick(data)
                        await self._send_to_kafka(tick_data, 'binance')
                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        logger.error(f"Binance WebSocket error: {ws.exception()}")
                        break
                        
    def _normalize_binance_tick(self, data):
        """Chuẩn hóa data từ Binance format"""
        return {
            'exchange': 'binance',
            'symbol': data['s'],
            'price': float(data['p']),
            'quantity': float(data['q']),
            'timestamp': data['T'],
            'is_buyer_maker': data['m'],
            'trade_id': data['t'],
            'ingest_time': datetime.utcnow().isoformat()
        }
        
    async def _send_to_kafka(self, tick_data, exchange):
        """Gửi tick data vào Kafka với error handling"""
        try:
            topic = f"crypto-tick-{tick_data['symbol'].lower()}"
            key = f"{exchange}:{tick_data['symbol']}"
            
            future = self.producer.send(
                topic,
                value=tick_data,
                key=key
            )
            
            # Wait for acknowledgment với timeout
            record_metadata = await asyncio.wrap_future(future)
            logger.debug(f"Sent to {record_metadata.topic}:{record_metadata.partition}:{record_metadata.offset}")
            
        except KafkaError as e:
            logger.error(f"Failed to send message to Kafka: {e}")
            # Implement dead letter queue hoặc fallback ở đây
            
    async def run(self):
        """Main loop - kết nối tất cả exchanges"""
        await self.initialize()
        self.running = True
        
        tasks = []
        for exchange_name, config in self.exchanges.items():
            for symbol in config['symbols']:
                if exchange_name == 'binance':
                    tasks.append(self.connect_binance(symbol))
                    
        logger.info(f"Starting {len(tasks)} WebSocket connections...")
        await asyncio.gather(*tasks, return_exceptions=True)
        
    async def shutdown(self):
        """Graceful shutdown"""
        self.running = False
        if self.producer:
            self.producer.flush()
            self.producer.close()
            logger.info("Kafka producer closed")

if __name__ == '__main__':
    producer = CryptoTickProducer(bootstrap_servers='localhost:9092')
    
    try:
        asyncio.run(producer.run())
    except KeyboardInterrupt:
        asyncio.run(producer.shutdown())

Consumer: Xử Lý Tick Data Và Tính Toán Indicators

Consumer là phần quan trọng nhất - nơi bạn xử lý tick data để tính toán indicators, execute trades, hoặc feed ML models:

from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json
import logging
from datetime import datetime, timedelta
from collections import defaultdict
import threading

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TickDataAggregator:
    """Aggregator để tính OHLCV từ tick data theo thời gian thực"""
    
    def __init__(self, interval_seconds=60):
        self.interval = timedelta(seconds=interval_seconds)
        self.candles = defaultdict(lambda: {
            'open': None,
            'high': float('-inf'),
            'low': float('inf'),
            'close': None,
            'volume': 0.0,
            'tick_count': 0
        })
        self.current_period_start = {}
        
    def process_tick(self, tick_data):
        """Xử lý một tick và update candle nếu cần"""
        symbol = tick_data['symbol']
        price = tick_data['price']
        quantity = tick_data['quantity']
        timestamp_ms = tick_data['timestamp']
        
        # Chuyển timestamp ms sang datetime
        current_time = datetime.fromtimestamp(timestamp_ms / 1000)
        period_start = current_time.replace(
            second=0, microsecond=0
        )
        
        # Check nếu cần tạo candle mới
        if symbol not in self.current_period_start:
            self.current_period_start[symbol] = period_start
            
        if period_start > self.current_period_start[symbol]:
            # Emit completed candle và reset
            completed_candle = self._emit_candle(symbol)
            self.current_period_start[symbol] = period_start
            self.candles[symbol] = {
                'open': None, 'high': float('-inf'),
                'low': float('inf'), 'close': None,
                'volume': 0.0, 'tick_count': 0
            }
            
        # Update current candle
        candle = self.candles[symbol]
        if candle['open'] is None:
            candle['open'] = price
            
        candle['high'] = max(candle['high'], price)
        candle['low'] = min(candle['low'], price)
        candle['close'] = price
        candle['volume'] += quantity
        candle['tick_count'] += 1
        
        return candle
        
    def _emit_candle(self, symbol):
        """Trả về completed candle"""
        candle = self.candles[symbol]
        return {
            'symbol': symbol,
            'open': candle['open'],
            'high': candle['high'],
            'low': candle['low'],
            'close': candle['close'],
            'volume': candle['volume'],
            'tick_count': candle['tick_count']
        }


class TickDataConsumer:
    """High-performance consumer với multi-threading"""
    
    def __init__(self, bootstrap_servers='localhost:9092'):
        self.bootstrap_servers = bootstrap_servers
        self.consumer = None
        self.running = False
        self.aggregator = TickDataAggregator(interval_seconds=60)
        self.stats = {
            'messages_processed': 0,
            'errors': 0,
            'last_message_time': None
        }
        self.stats_lock = threading.Lock()
        
    def initialize(self):
        """Khởi tạo consumer với các config tối ưu cho high-throughput"""
        try:
            self.consumer = KafkaConsumer(
                'crypto-tick-btcusdt',
                bootstrap_servers=self.bootstrap_servers,
                group_id='tick-processor-group',
                auto_offset_reset='latest',
                enable_auto_commit=False,
                max_poll_records=500,
                fetch_max_bytes=1048576,  # 1MB
                max_partition_fetch_bytes=1048576,
                session_timeout_ms=30000,
                heartbeat_interval_ms=10000
            )
            logger.info("Kafka consumer initialized successfully")
            return True
        except KafkaError as e:
            logger.error(f"Failed to initialize consumer: {e}")
            return False
            
    def process_messages(self):
        """Xử lý messages với batch processing"""
        try:
            while self.running:
                # Poll với timeout ngắn để không block
                message_batch = self.consumer.poll(timeout_ms=100)
                
                if not message_batch:
                    continue
                    
                for topic_partition, messages in message_batch.items():
                    for message in messages:
                        try:
                            tick_data = json.loads(message.value.decode('utf-8'))
                            candle = self.aggregator.process_tick(tick_data)
                            
                            with self.stats_lock:
                                self.stats['messages_processed'] += 1
                                self.stats['last_message_time'] = datetime.now()
                                
                            # Log candle update mỗi 1000 ticks
                            if self.stats['messages_processed'] % 1000 == 0:
                                logger.info(
                                    f"Processed {self.stats['messages_processed']} ticks | "
                                    f"Current BTC price: {candle.get('close')} | "
                                    f"Volume: {candle.get('volume', 0):.4f}"
                                )
                                
                        except json.JSONDecodeError as e:
                            logger.error(f"Invalid JSON: {e}")
                            with self.stats_lock:
                                self.stats['errors'] += 1
                        except Exception as e:
                            logger.error(f"Error processing message: {e}")
                            with self.stats_lock:
                                self.stats['errors'] += 1
                                
                # Commit offsets sau khi xử lý batch thành công
                self.consumer.commit()
                
        except Exception as e:
            logger.error(f"Consumer error: {e}")
        finally:
            if self.consumer:
                self.consumer.close()
                
    def start(self):
        """Start consumer trong background thread"""
        if not self.initialize():
            return False
            
        self.running = True
        self.consumer_thread = threading.Thread(
            target=self.process_messages,
            daemon=True
        )
        self.consumer_thread.start()
        logger.info("Consumer started in background thread")
        return True
        
    def stop(self):
        """Stop consumer gracefully"""
        self.running = False
        if hasattr(self, 'consumer_thread'):
            self.consumer_thread.join(timeout=5)
        logger.info(f"Final stats: {self.stats}")
        
    def get_stats(self):
        """Get current statistics"""
        with self.stats_lock:
            return self.stats.copy()


if __name__ == '__main__':
    consumer = TickDataConsumer(bootstrap_servers='localhost:9092')
    
    try:
        consumer.start()
        # Keep main thread alive
        while True:
            import time
            time.sleep(10)
            stats = consumer.get_stats()
            logger.info(f"Stats update: {stats['messages_processed']} processed, {stats['errors']} errors")
    except KeyboardInterrupt:
        consumer.stop()

Streaming Với Kafka Streams Cho Real-time Analytics

Để xử lý phức tạp hơn như windowed aggregations, joins giữa multiple streams, Kafka Streams là lựa chọn tối ưu:

# requirements: pip install kafka-streams

from kafka_streams import KafkaStreams
import json
import logging
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class VolumeAlertStreamProcessor:
    """
    Xử lý stream để detect volume spikes và trigger alerts.
    Sử dụng time-windowed aggregation với late arrival handling.
    """
    
    def __init__(self, bootstrap_servers='localhost:9092'):
        self.bootstrap_servers = bootstrap_servers
        self.streams = None
        self.alert_threshold = 10.0  # Volume spike x10 so với average
        
    def build_topology(self):
        """Định nghĩa processing topology"""
        streams = KafkaStreams(
            bootstrap_servers=self.bootstrap_servers,
            application_id='volume-alert-processor',
            state_dir='/tmp/kafka-streams',
            processing_guarantee='exactly_once_v2'
        )
        
        # Define time window: 5 minutes tumbling window
        windowed_volume = (
            streams
            .stream('crypto-tick-btcusdt')
            .map_values(lambda v: json.loads(v))
            .filter(lambda k, v: v.get('exchange') == 'binance')
            .group_by_key()
            .windowed_by(
                TimeWindows.of(timedelta(minutes=5))
                    .grace(timedelta(seconds=30))  # Late arrival grace period
            )
            .aggregate(
                initializer=lambda: {
                    'total_volume': 0.0,
                    'tick_count': 0,
                    'prices': [],
                    'start_time': None
                },
                aggregator=lambda k, v, agg: {
                    'total_volume': agg['total_volume'] + v.get('quantity', 0),
                    'tick_count': agg['tick_count'] + 1,
                    'prices': agg['prices'] + [v.get('price', 0)],
                    'start_time': agg['start_time'] or v.get('timestamp')
                },
                serde=JsonSerdes()
            )
        )
        
        # Detect volume spikes
        alerts = (
            windowed_volume
            .suppress(Suppressed.until_window_close(BufferConfig.unbounded()))
            .filter(lambda k, v: v['total_volume'] > self.alert_threshold * self._get_average_volume())
            .map_values(lambda v: {
                'symbol': 'BTCUSDT',
                'window_start': datetime.fromtimestamp(v['start_time'] / 1000).isoformat() if v['start_time'] else None,
                'total_volume': v['total_volume'],
                'tick_count': v['tick_count'],
                'avg_price': sum(v['prices']) / len(v['prices']) if v['prices'] else 0,
                'alert_type': 'VOLUME_SPIKE',
                'severity': 'HIGH' if v['total_volume'] > 50 else 'MEDIUM'
            })
        )
        
        # Output alerts to different topics based on severity
        alerts
            .filter(lambda k, v: v['severity'] == 'HIGH')
            .to('crypto-alerts-high')
            
        alerts
            .filter(lambda k, v: v['severity'] == 'MEDIUM')
            .to('crypto-alerts-medium')
            
        # Also output to monitoring dashboard
        alerts
            .peek(lambda k, v: logger.info(f"ALERT: {v}"))
            
        return streams
        
    def _get_average_volume(self):
        """Lấy average volume từ cache - implement với Redis hoặc local cache"""
        # Placeholder - trong production nên lấy từ Redis/state store
        return 1.0
        
    def start(self):
        """Start Kafka Streams application"""
        self.streams = self.build_topology()
        
        # Set up state listener for debugging
        self.streams.set_state_listener(
            lambda new_state, old_state: 
                logger.info(f"State transition: {old_state} -> {new_state}")
        )
        
        self.streams.start()
        logger.info("Kafka Streams application started")
        
    def stop(self):
        """Stop gracefully"""
        if self.streams:
            self.streams.close()
            logger.info("Kafka Streams application stopped")



Alternative: Sử dụng Faust (pure Python alternative to Kafka Streams)


pip install faust


import faust

app = faust.App(
    'crypto-tick-processor',
    broker='kafka://localhost:9092',
    value_serializer='json'
)

tick_topic = app.topic('crypto-tick-btcusdt', value_type=dict)
alert_topic = app.topic('crypto-alerts', value_type=dict)


Tumbling window aggregation

windowed = tick_topic.groupby('symbol').hopeful_window(
    size=timedelta(minutes=5),
    step=timedelta(minutes=1)
)

@app.agent(windowed)
async def process_volume_alerts(window):
    async for key, events in window.items():
        total_volume = sum(e.get('quantity', 0) for e in events)
        tick_count = len(events)
        
        if total_volume > 10:  # Threshold
            await alert_topic.send(
                value={
                    'symbol': key,
                    'window_start': window.start,
                    'window_end': window.end,
                    'total_volume': total_volume,
                    'tick_count': tick_count,
                    'alert_type': 'VOLUME_SPIKE'
                }
            )

if __name__ == '__main__':
    app.start()

Tối Ưu Hóa Hiệu Suất Cho High-Frequency Data

Để đạt được throughput cao nhất với Kafka, đây là những config quan trọng tôi đã tune qua nhiều lần thử nghiệm:

# producer_optimized.py - Tối ưu producer cho throughput cao nhất

from kafka import KafkaProducer
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OptimizedTickProducer:
    """
    Producer với các optimizations cho high-throughput tick data:
    - Batching với linger.ms
    - Compression
    - Async send với callback
    - Partitioning strategy
    """
    
    def __init__(self, bootstrap_servers='localhost:9092'):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self.pending_futures = []
        
    def create_optimized_producer(self):
        """
        Các config quan trọng cần tối ưu:
        - batch.size: Tăng để batch nhiều messages hơn trước khi send
        - linger.ms: Đợi thêm ms trước khi send batch (tăng throughput, tăng latency)
        - buffer.memory: Tăng buffer cho producer
        - compression.type: gzip/snappy/lz4 zstd (tiết kiệm bandwidth)
        - acks: 1 cho throughput cao, 'all' cho durability
        """
        return KafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            
            # Batching optimizations
            batch_size=65536,  # 64KB (tăng từ default 16KB)
            linger_ms=10,  # Đợi 10ms để batch messages
            
            # Buffer optimizations  
            buffer_memory=67108864,  # 64MB buffer
            max_request_size=1048576,  # 1MB max message
            
            # Compression
            compression_type='lz4',  # LZ4: fast compression, good ratio
            
            # Reliability
            acks=1,  # Leader acknowledgment only - trade-off speed vs durability
            retries=3,
            max_in_flight_requests_per_connection=5,  # Allow concurrent batches
            
            # Serialization
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None,
            
            # Performance
            connections_max_idle_ms=60000,
            request_timeout_ms=30000,
        )
        
    def send_with_callback(self, topic, key, value):
        """Send message với callback để track thành công"""
        future = self.producer.send(topic, key=key, value=value)
        
        # Add callback để handle errors asynchronously
        future.add_callback(
            lambda m: logger.debug(
                f"Sent to {m.topic}:{m.partition}:{m.offset}"
            )
        )
        future.add_errback(
            lambda e: logger.error(f"Send failed: {e}")
        )
        
        self.pending_futures.append(future)
        
        # Flush periodically để tránh memory leak
        if len(self.pending_futures) >= 1000:
            self._flush_pending()
            
    def _flush_pending(self):
        """Flush pending futures và wait for completion"""
        for future in self.pending_futures:
            try:
                future.get(timeout=10)
            except Exception as e:
                logger.error(f"Future failed: {e}")
        self.pending_futures.clear()
        
    def close(self):
        """Graceful shutdown - đợi tất cả messages được send"""
        logger.info(f"Flushing {len(self.pending_futures)} pending messages...")
        self._flush_pending()
        self.producer.flush()
        self.producer.close()
        logger.info("Producer closed")



Consumer optimizations

class OptimizedTickConsumer:
    """
    Consumer với optimizations:
    - Fetch optimization
    - Parallel processing
    - Auto commit tuning
    """
    
    @staticmethod
    def create_optimized_consumer():
        """
        Các config quan trọng:
        - fetch.min.bytes: Minimum bytes to fetch
        - fetch.max.wait.ms: Max time to wait for fetch
        - max.poll.records: Records per poll
        - max.partition.fetch.bytes: Bytes per partition
        """
        return

Tài nguyên liên quan
Bài viết liên quan