Полиглотные Event-Driven системы с Kafka, RabbitMQ и gRPC на Java, Go и Node.js

const { Kafka } = require('kafkajs');
 
class OrderEventConsumer {
  constructor() {
    this.kafka = new Kafka({
      clientId: 'notification-service',
      brokers: ['localhost:9092'],
      // Retry стратегия - важна для переподключений
      retry: {
        initialRetryTime: 100,
        retries: 8
      }
    });
    
    this.consumer = this.kafka.consumer({ 
      groupId: 'notification-group',
      // Читаем с самого старого непрочитанного сообщения
      fromBeginning: false,
      // Автоматический commit offset - опасная штука
      autoCommit: false
    });
  }
 
  async start() {
    await this.consumer.connect();
    await this.consumer.subscribe({ 
      topic: 'order-events',
      // Подписываемся на все партиции топика
      fromBeginning: false 
    });
 
    await this.consumer.run({
      // Обрабатываем сообщения пачками для производительности
      eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning }) => {
        for (let message of batch.messages) {
          if (!isRunning()) break; // Проверяем, не останавливается ли консюмер
          
          try {
            const event = JSON.parse(message.value.toString());
            await this.processEvent(event);
            
            // Коммитим offset только после успешной обработки
            await resolveOffset(message.offset);
            await heartbeat(); // Сигнализируем Kafka, что мы живы
          } catch (error) {
            console.error(`Error processing message at offset ${message.offset}:`, error);
            // Тут решаем: пропустить сообщение или упасть
            // В production лучше отправить в DLQ (Dead Letter Queue)
            throw error; // Останавливаем обработку батча
          }
        }
      }
    });
  }
 
  async processEvent(event) {
    // Имитация отправки уведомления
    console.log(`Processing ${event.eventType} for order ${event.orderId}`);
    
    // Идемпотентность на стороне консюмера
    const alreadyProcessed = await this.checkIfProcessed(event.orderId, event.eventType);
    if (alreadyProcessed) {
      console.log(`Event already processed, skipping`);
      return;
    }
    
    // Отправляем уведомление пользователю
    await this.sendNotification(event);
    
    // Сохраняем факт обработки
    await this.markAsProcessed(event.orderId, event.eventType);
  }
  
  // Заглушки для примера
  async checkIfProcessed(orderId, eventType) {
    // В реальности - проверка в Redis или БД
    return false;
  }
  
  async markAsProcessed(orderId, eventType) {
    // Сохраняем в Redis с TTL равным retention периоду Kafka
  }
  
  async sendNotification(event) {
    // Отправка через WebSocket, email, push
  }
}

async processWithRetry(event, maxRetries = 3) {
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      await this.processEvent(event);
      return; // Успех
    } catch (error) {
      if (this.isTransientError(error) && attempt < maxRetries) {
        // Ждем с экспоненциальной задержкой
        const delay = Math.pow(2, attempt) * 100;
        await new Promise(resolve => setTimeout(resolve, delay));
        console.log(`Retry attempt ${attempt} after ${delay}ms`);
      } else {
        // Перманентная ошибка или исчерпаны попытки
        await this.sendToDLQ(event, error);
        return; // Не бросаем исключение, чтобы не останавливать батч
      }
    }
  }
}
 
isTransientError(error) {
  // Проверяем код ошибки
  return error.code === 'ECONNREFUSED' || 
         error.code === 'ETIMEDOUT' ||
         error.message.includes('temporary');
}

const amqp = require('amqplib');
 
class EmailWorker {
  async start() {
    this.connection = await amqp.connect('amqp://localhost');
    this.channel = await this.connection.createChannel();
 
    // Prefetch 1 - строго последовательная обработка
    await this.channel.prefetch(1);
 
    await this.channel.consume('email.queue', async (msg) => {
      if (!msg) return;
 
      try {
        const notification = JSON.parse(msg.content.toString());
        await this.sendEmail(notification);
        
        // Подтверждаем после успешной отправки
        this.channel.ack(msg);
      } catch (error) {
        console.error('Email sending failed:', error);
        
        // Проверяем счетчик попыток
        const retryCount = (msg.properties.headers['x-retry-count'] || 0);
        
        if (retryCount < 3) {
          // Переотправляем с задержкой
          await this.retryWithDelay(msg, retryCount + 1);
          this.channel.ack(msg); // Убираем из основной очереди
        } else {
          // Отправляем в DLQ
          this.channel.nack(msg, false, false);
        }
      }
    });
  }
 
  async retryWithDelay(msg, retryCount) {
    // Переотправляем в ту же очередь через delay exchange
    const delayMs = Math.pow(2, retryCount) * 1000; // Exponential backoff
    
    this.channel.publish(
      'delayed.exchange',
      'email.queue',
      msg.content,
      {
        headers: { 
          'x-retry-count': retryCount,
          'x-delay': delayMs 
        }
      }
    );
  }
 
  async sendEmail(notification) {
    // Реальная отправка через SMTP или API
    console.log(`Sending email to ${notification.userId}`);
    await new Promise(resolve => setTimeout(resolve, 500));
  }
}

class RetryableConsumer {
  async setupQueues() {
    const channel = await this.connection.createChannel();
 
    // Основная очередь с маршрутизацией в retry exchange
    await channel.assertQueue('payments.process', {
      durable: true,
      arguments: {
        'x-dead-letter-exchange': 'payments.retry'
      }
    });
 
    // Retry exchange
    await channel.assertExchange('payments.retry', 'direct', { durable: true });
 
    // Три retry очереди с разными TTL
    const retryDelays = [1000, 5000, 15000]; // 1s, 5s, 15s
    
    for (let i = 0; i < retryDelays.length; i++) {
      const queueName = `payments.retry.${i}`;
      
      await channel.assertQueue(queueName, {
        durable: true,
        arguments: {
          'x-message-ttl': retryDelays[i],
          'x-dead-letter-exchange': '', // Default exchange
          'x-dead-letter-routing-key': 'payments.process'
        }
      });
 
      await channel.bindQueue(queueName, 'payments.retry', [INLINE]retry.${i}[/INLINE]);
    }
 
    // DLQ для окончательно битых сообщений
    await channel.assertQueue('payments.dlq', { durable: true });
  }
 
  async consume() {
    const channel = await this.connection.createChannel();
    await channel.prefetch(1);
 
    channel.consume('payments.process', async (msg) => {
      if (!msg) return;
 
      const retryCount = msg.properties.headers['x-retry-count'] || 0;
 
      try {
        await this.processPayment(JSON.parse(msg.content.toString()));
        channel.ack(msg);
      } catch (error) {
        console.error(`Payment processing failed (attempt ${retryCount + 1}):`, error);
 
        if (retryCount < 3) {
          // Отправляем в соответствующую retry очередь
          channel.publish(
            'payments.retry',
            [INLINE]retry.${retryCount}[/INLINE],
            msg.content,
            {
              headers: { 'x-retry-count': retryCount + 1 },
              persistent: true
            }
          );
          channel.ack(msg); // Убираем из основной очереди
        } else {
          // После трех попыток - в DLQ
          channel.publish('', 'payments.dlq', msg.content, {
            headers: { 
              'x-retry-count': retryCount,
              'x-error': error.message 
            },
            persistent: true
          });
          channel.ack(msg);
        }
      }
    });
  }
}

const axios = require('axios');
 
async function collectMetrics() {
  const response = await axios.get('http://localhost:15672/api/queues', {
    auth: { username: 'guest', password: 'guest' }
  });
 
  for (const queue of response.data) {
    console.log(`Queue: ${queue.name}`);
    console.log(`  Messages: ${queue.messages}`);
    console.log(`  Rate: ${queue.messages_details.rate}/s`);
    console.log(`  Unacked: ${queue.messages_unacknowledged}`);
    console.log(`  Consumers: ${queue.consumers}`);
 
    // Алерт если очередь разрастается
    if (queue.messages > 10000) {
      sendAlert(`Queue ${queue.name} has ${queue.messages} messages!`);
    }
 
    // Алерт если консюмеров нет
    if (queue.consumers === 0 && queue.messages > 0) {
      sendAlert(`Queue ${queue.name} has no consumers!`);
    }
  }
}
 
setInterval(collectMetrics, 30000); // Каждые 30 секунд

const grpc = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');
const { promisify } = require('util');
 
class OrderClient {
  constructor(serverAddress) {
    const packageDefinition = protoLoader.loadSync('order.proto', {
      keepCase: true,
      longs: String,
      enums: String,
      defaults: true,
      oneofs: true
    });
 
    const proto = grpc.loadPackageDefinition(packageDefinition).orderservice;
    
    this.client = new proto.OrderService(
      serverAddress,
      grpc.credentials.createInsecure()
    );
 
    // Промисифицируем методы
    this.createOrder = promisify(this.client.createOrder.bind(this.client));
    this.getOrderStatus = promisify(this.client.getOrderStatus.bind(this.client));
  }
 
  async placeOrder(userId, items, address) {
    const request = {
      user_id: userId,
      items: items,
      delivery_address: address,
      payment_method: 'CREDIT_CARD'
    };
 
    try {
      const response = await this.createOrder(request, {
        deadline: Date.now() + 5000  // 5 секунд
      });
 
      console.log(`Order placed: ${response.order_id}`);
      return response;
 
    } catch (error) {
      if (error.code === grpc.status.DEADLINE_EXCEEDED) {
        console.error('Request timeout');
      } else if (error.code === grpc.status.INVALID_ARGUMENT) {
        console.error(`Invalid request: ${error.details}`);
      }
      throw error;
    }
  }
 
  async checkStatus(orderId) {
    try {
      const status = await this.getOrderStatus({ order_id: orderId });
      return status;
    } catch (error) {
      if (error.code === grpc.status.NOT_FOUND) {
        throw new Error(`Order ${orderId} not found`);
      }
      throw error;
    }
  }
}

async function uploadOrders(filePath) {
  const call = client.batchCreateOrders((error, response) => {
    if (error) {
      console.error('Upload failed:', error);
    } else {
      console.log(`Uploaded ${response.created_count} orders`);
    }
  });
 
  const stream = fs.createReadStream(filePath);
  const rl = readline.createInterface({ input: stream });
 
  for await (const line of rl) {
    const order = parseCSVLine(line);
    
    call.write({
      user_id: order.userId,
      items: order.items,
      delivery_address: order.address
    });
    
    // Backpressure - ждем, если буфер полон
    if (!call.writeable) {
      await new Promise(resolve => call.once('drain', resolve));
    }
  }
 
  call.end(); // Сигнализируем окончание стрима
}

function createLoggingInterceptor() {
  return function(options, nextCall) {
    return new InterceptingCall(nextCall(options), {
      start: function(metadata, listener, next) {
        const methodName = options.method_definition.path;
        const startTime = Date.now();
        
        console.log(`---> ${methodName} started`);
 
        const newListener = {
          onReceiveStatus: function(status, next) {
            const duration = Date.now() - startTime;
            
            if (status.code === grpc.status.OK) {
              console.log(`<--- ${methodName} completed in ${duration}ms`);
            } else {
              console.error(`<--- ${methodName} failed in ${duration}ms: ${status.details}`);
            }
            
            next(status);
          }
        };
 
        next(metadata, newListener);
      }
    });
  };
}
 
// Применяем к клиенту
const client = new proto.OrderService(
  serverAddress,
  grpc.credentials.createInsecure(),
  { interceptors: [createLoggingInterceptor()] }
);

// Плохо - упадет на неизвестном поле при strict mode
const user = JSON.parse(data);
console.log(user.id, user.name);
 
// Хорошо - игнорируем лишнее, работаем с нужным
const user = JSON.parse(data);
const { id, name } = user;
if (id && name) {
  processUser(id, name);
}

const { trace } = require('@opentelemetry/api');
 
async function handleRequest(req, res) {
  const tracer = trace.getTracer('api-service');
  
  return tracer.startActiveSpan('handle_request', async (span) => {
    span.setAttribute('http.method', req.method);
    span.setAttribute('http.url', req.url);
    
    try {
      const result = await processRequest(req);
      res.json(result);
    } catch (error) {
      span.recordException(error);
      span.setStatus({ code: SpanStatusCode.ERROR });
      throw error;
    } finally {
      span.end();
    }
  });
}

const express = require('express');
const { Kafka } = require('kafkajs');
 
class APIGateway {
  constructor() {
    this.app = express();
    this.app.use(express.json());
    
    this.kafka = new Kafka({
      clientId: 'api-gateway',
      brokers: ['kafka:9092'],
      retry: { retries: 5 }
    });
    
    this.producer = this.kafka.producer({
      idempotent: true,
      maxInFlightRequests: 1
    });
    
    this.isShuttingDown = false;
  }
 
  async start() {
    await this.producer.connect();
    
    this.app.post('/orders', async (req, res) => {
      if (this.isShuttingDown) {
        return res.status(503).json({ error: 'Service shutting down' });
      }
 
      const { userId, items, address } = req.body;
      const orderId = `ORD-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
 
      const event = {
        eventType: 'OrderCreated',
        orderId,
        userId,
        items,
        address,
        timestamp: Date.now()
      };
 
      try {
        await this.producer.send({
          topic: 'order-events',
          messages: [{
            key: orderId,
            value: JSON.stringify(event)
          }]
        });
 
        res.status(201).json({ orderId, status: 'pending' });
      } catch (error) {
        console.error('Failed to publish event:', error);
        res.status(500).json({ error: 'Order processing failed' });
      }
    });
 
    this.server = this.app.listen(3000, () => {
      console.log('API Gateway listening on :3000');
    });
 
    this.setupGracefulShutdown();
  }
 
  setupGracefulShutdown() {
    const shutdown = async (signal) => {
      console.log(`${signal} received, starting graceful shutdown...`);
      this.isShuttingDown = true;
 
      // Перестаем принимать новые запросы
      this.server.close(() => {
        console.log('HTTP server closed');
      });
 
      // Флашим все pending сообщения
      await this.producer.disconnect();
      console.log('Kafka producer disconnected');
 
      process.exit(0);
    };
 
    process.on('SIGTERM', () => shutdown('SIGTERM'));
    process.on('SIGINT', () => shutdown('SIGINT'));
  }
}
 
const gateway = new APIGateway();
gateway.start().catch(console.error);

import http from 'k6/http';
import { check, sleep } from 'k6';
 
export const options = {
  stages: [
    { duration: '1m', target: 50 },   // Разгон до 50 пользователей
    { duration: '3m', target: 50 },   // Держим нагрузку
    { duration: '1m', target: 100 },  // Пик
    { duration: '2m', target: 100 },  // Держим пик
    { duration: '1m', target: 0 },    // Спад
  ],
  thresholds: {
    http_req_duration: ['p(95)<500'], // 95% запросов быстрее 500ms
    http_req_failed: ['rate<0.01'],   // Меньше 1% ошибок
  },
};
 
export default function() {
  const payload = JSON.stringify({
    userId: [INLINE]user-${__VU}-${__ITER}[/INLINE],
    items: [
      { productId: 'prod-1', quantity: 2, price: 29.99 },
      { productId: 'prod-2', quantity: 1, price: 49.99 },
    ],
    address: {
      street: '123 Main St',
      city: 'Test City',
      postalCode: '12345',
      country: 'US'
    }
  });
 
  const params = {
    headers: { 'Content-Type': 'application/json' },
  };
 
  const res = http.post('http://localhost:3000/orders', payload, params);
 
  check(res, {
    'status is 201': (r) => r.status === 201,
    'response has orderId': (r) => JSON.parse(r.body).orderId !== undefined,
  });
 
  sleep(1);
}

const promClient = require('prom-client');
 
const register = new promClient.Registry();
promClient.collectDefaultMetrics({ register });
 
const ordersCreated = new promClient.Counter({
  name: 'orders_created_total',
  help: 'Total orders created',
  registers: [register]
});
 
const orderCreationDuration = new promClient.Histogram({
  name: 'order_creation_duration_seconds',
  help: 'Order creation duration',
  buckets: [0.1, 0.5, 1, 2, 5],
  registers: [register]
});
 
// В обработчике
app.post('/orders', async (req, res) => {
  const end = orderCreationDuration.startTimer();
  
  try {
    // ... создание заказа
    ordersCreated.inc();
  } finally {
    end();
  }
});
 
// Endpoint для Prometheus
app.get('/metrics', (req, res) => {
  res.set('Content-Type', register.contentType);
  res.end(register.metrics());
});