// ILP形式でのデータ送信
let line = format!(
    "trades,exchange={},symbol={},side={},trade_id={} price={},size={},sequence={},quality_score={} {}",
    escape_tag(&trade.exchange),
    escape_tag(&trade.symbol),
    escape_tag(&trade.side),
    escape_tag(&trade.trade_id),
    trade.price,
    trade.size,
    trade.sequence.unwrap_or(0),
    trade.quality_score,
    trade.timestamp.timestamp_nanos_opt().unwrap_or(0)
);

pub struct IlpClient {
    connection: Arc<Mutex<TcpStream>>,
    connection_pool: Arc<DashMap<String, ConnectionInfo>>,
    // ...
}

pub fn optimal_batch_size(&self) -> usize {
    let throughput_based_size = match state.current_throughput as u32 {
        0..=50 => self.config.min_batch_size,
        51..=100 => 50,
        101..=200 => 100,
        201..=500 => 200,
        501..=1000 => 500,
        _ => self.config.max_batch_size,
    };

    // バックプレッシャーに基づく調整
    match state.backpressure_level {
        BackpressureLevel::None => throughput_based_size,
        BackpressureLevel::Low => (throughput_based_size as f64 * 0.9) as usize,
        BackpressureLevel::Medium => (throughput_based_size as f64 * 0.7) as usize,
        BackpressureLevel::High => (throughput_based_size as f64 * 0.5) as usize,
    }
}

// キューの深さに基づくバックプレッシャー
if *max_depth > 800 && !backpressure_active {
    info!("Activating backpressure, max queue depth: {}", max_depth);
    consumer.pause(&partitions)?;
    backpressure_active = true;
} else if *max_depth < 200 && backpressure_active {
    info!("Deactivating backpressure, max queue depth: {}", max_depth);
    consumer.resume(&partitions)?;
    backpressure_active = false;
}

pub struct ParallelProcessor {
    workers: Vec<Worker>,
    dispatcher: Dispatcher,
    config: ProcessorConfig,
}

// ラウンドロビンによるタスク分配
fn dispatch(&self, item: WorkItem) -> Result<()> {
    let mut index = self.round_robin_index.lock();
    let worker_idx = *index % self.senders.len();
    *index = (*index + 1) % self.senders.len();
    self.send_to_worker(worker_idx, item)
}

// Bytesを使用したゼロコピー
if let Some(payload) = msg.payload() {
    let data = Bytes::copy_from_slice(payload);
    processor.process(data).await?;
}

// ゼロコピーデシリアライゼーション
let trade: NormalizedTrade = serde_json::from_slice(&data)?;

lazy_static! {
    static ref TRADES_PROCESSED: prometheus::Counter = 
        prometheus::Counter::new("trades_processed_total", "Total trades processed");

    static ref PROCESSING_LATENCY: prometheus::Histogram = 
        prometheus::Histogram::with_opts(
            prometheus::HistogramOpts::new("processing_latency_seconds", "Trade processing latency")
        );
}

# 最適化ビルド
cargo build --release --bin consumer_optimized

# 実行（環境変数設定）
RUST_LOG=info \
KAFKA_BROKERS=redpanda:9092 \
QUESTDB_HOST=questdb \
QUESTDB_ILP_PORT=9000 \
BATCH_SIZE=1000 \
NUM_WORKERS=4 \
cargo run --release --bin consumer_optimized

# Dockerfile.coinbase-consumer-optimized
FROM rust:1.81-slim as builder
WORKDIR /app

# 依存関係のキャッシュ
COPY rust/Cargo.toml rust/Cargo.lock ./
COPY rust/coinbase-consumer ./coinbase-consumer
RUN cargo build --release --bin consumer_optimized

# 実行イメージ
FROM debian:bookworm-slim
COPY --from=builder /app/target/release/consumer_optimized /usr/local/bin/
CMD ["consumer_optimized"]

Failed to connect to QuestDB ILP endpoint

Activating backpressure, max queue depth: 850

# docker-compose.yml
coinbase-consumer-optimized:
  environment:
    - RUST_LOG=coinbase_consumer=info
    - KAFKA_BROKERS=redpanda:9092
    - QUESTDB_HOST=questdb
    - QUESTDB_ILP_PORT=9000
    - BATCH_SIZE=1000
    - NUM_WORKERS=8
    - KAFKA_GROUP_ID=coinbase-consumer-optimized
  deploy:
    resources:
      limits:
        cpus: '2'
        memory: 512M
      reservations:
        cpus: '1'
        memory: 256M

# Prometheusアラートルール例
groups:
  - name: consumer_alerts
    rules:
      - alert: HighProcessingLatency
        expr: processing_latency_seconds > 0.1
        for: 5m
        annotations:
          summary: "High processing latency detected"

      - alert: LowThroughput
        expr: rate(trades_processed_total[5m]) < 100
        for: 10m
        annotations:
          summary: "Low throughput detected"