Apache Camel и Apache Kafka - в чем разница?

from("file:data/inbox")
    .filter().xpath("/order[@test='true']")
    .to("jms:queue:TEST.ORDERS");
 
from("file:data/inbox")
    .filter().xpath("/order[@test='false']")
    .to("jms:queue:PRODUCTION.ORDERS");

Топик "orders"
  ├── Партиция 0: [m0, m1, m2, m3, m4, ...]
  ├── Партиция 1: [m0, m1, m2, ...]
  └── Партиция 2: [m0, m1, m2, m3, ...]

// Компонент для работы с HTTP
camelContext.addComponent("http", new HttpComponent());
// Компонент для работы с JMS
camelContext.addComponent("jms", new JmsComponent());

KStream<String, Order> orders = builder.stream("orders-topic");
KStream<String, OrderValidated> validatedOrders = orders
    .filter((key, order) -> order.getAmount() > 0)
    .mapValues(order -> new OrderValidated(order));
validatedOrders.to("validated-orders-topic");

jms:queue:orders?transacted=true&concurrentConsumers=5

from("jms:queue:orders")
    .errorHandler(deadLetterChannel("jms:queue:failed")
        .maximumRedeliveries(3)
        .redeliveryDelay(1000)
        .backOffMultiplier(2))
    .to("bean:orderProcessor");

from("file:inbox?noop=true")
  .unmarshal().csv()
  .split(body())
  .marshal().json()
  .to("kafka:converted-data");

Properties props = new Properties();
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
Producer<String, Customer> producer = new KafkaProducer<>(props);

final AvroSerializer<Customer> serializer = new AvroSerializer<>(
    url, false, 100, schemaRegistry, Customer.class);

PropertiesComponent pc = new PropertiesComponent();
pc.setLocation("classpath:config.properties");
camelContext.addComponent("properties", pc);
 
from("{{sourceSystems}}")
  .to("{{targetSystem}}");

bootstrap.servers=localhost:9092
acks=all
retries=0
buffer.memory=33554432

Топик "orders", коэффициент репликации = 3
├── Партиция 0:
│   ├── Реплика на Брокере 1 (лидер)
│   ├── Реплика на Брокере 2
│   └── Реплика на Брокере 3
├── Партиция 1:
│   ├── Реплика на Брокере 2 (лидер)
│   ├── Реплика на Брокере 3
│   └── Реплика на Брокере 1
└── ...

from("seda:orders?concurrentConsumers=10")
  .to("bean:processor");

Группа потребителей "order-processors"
├── Потребитель 1 ← Партиция 0
├── Потребитель 2 ← Партиция 1
└── Потребитель 3 ← Партиция 2

from("amqp:queue:orders")
  .throttle(100).timePeriodMillis(1000)
  .to("bean:orderProcessor");

props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);

from("direct:start")
  .wireTap("direct:metrics")
  .to("direct:business");
 
from("direct:metrics")
  .to("metrics:counter:messages?increment=1");

from("cxf:bean:accountingServiceEndpoint")
  .process(new SoapToPojoConverter())
  .choice()
    .when(header("customerType").isEqualTo("corporate"))
      .to("rest:post:customers/corporate")
    .otherwise()
      .to("rest:post:customers/individual")
  .end()
  .process(new ResponseFormatter())
  .to("grpc://documentService?method=storeDocument");

from("jt400:USER:PASSWORD@SYSTEM/LIBRARY/QUEUE?format=binary")
  .unmarshal().custom(ebcdicToAsciiConverter)
  .split().xpath("/records/record")
  .process(new LegacyRecordTransformer())
  .to("kafka:modernized-data");

@KafkaListener(topics = "device-telemetry", groupId = "anomaly-detection")
public void processMessage(ConsumerRecord<String, DeviceTelemetry> record) {
    DeviceTelemetry telemetry = record.value();
    if (isAnomaly(telemetry)) {
        alertService.sendAlert(telemetry.getDeviceId(), "Аномальное поведение устройства");
    }
    metricsRepository.save(telemetry);
}

Сервис заказов → Топик "order-created" → Сервис доставки
                                        → Сервис оплаты
                                        → Сервис уведомлений

KStream<String, Transaction> transactions = builder.stream("transactions");
KTable<String, Double> fraudScore = transactions
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .aggregate(
        () -> 0.0,
        (key, transaction, score) -> updateFraudScore(transaction, score),
        Materialized.with(Serdes.String(), Serdes.Double())
    );
fraudScore
    .filter((key, score) -> score > fraudThreshold)
    .toStream()
    .to("fraud-alerts");

from("file:orders?noop=true")
  .unmarshal().csv()
  .split(body())
  .process(new OrderEnricher())
  .to("kafka:enriched-orders");
 
from("kafka:processed-orders")
  .choice()
    .when(header("destination").isEqualTo("email"))
      .to("smtp:orders@company.com")
    .when(header("destination").isEqualTo("erp"))
      .marshal().json()
      .to("jms:queue:ERP.ORDERS")
    .otherwise()
      .to("log:fallback?level=WARN");

from("mqtt:sensor/+/temperature")
.transform(simple("{ \"id\": \"${header.mqtt.topic}\", \"value\": ${body}, \"timestamp\": ${date:now:yyyy-MM-dd'T'HH:mm:ss.SSSZ} }"))
.to("kafka:temperature-readings");

from("file:incoming/suppliers?include=.*\\.edi$")
.unmarshal().edi()
.split(xpath("/orders/order"))
.choice()
  .when(xpath("/order/status = 'SHIPPED'"))
    .to("bean:inventoryService?method=updateStockLevels")
  .when(xpath("/order/status = 'INVOICED'"))
    .to("bean:accountingService?method=createInvoice")
.end()
.process(new AcknowledgmentCreator())
.marshal().edi()
.to("file:outgoing/suppliers");

// Конфигурация Debezium в Camel для отслеживания изменений в MySQL
from("debezium-mysql:dbserver1?databaseHostname=localhost&databasePort=3306&databaseUser=debezium&databasePassword=dbz&databaseServerName=my-app-connector&databaseHistoryFileName=/tmp/dbhistory.dat&tableIncludeList=inventory.customers")
.to("kafka:mysql-customers-changed");

from("hl7-mllp://0.0.0.0:8888?requireAck=true")
.process(new PatientIdExtractor())
.choice()
  .when(header("messageType").isEqualTo("ADT"))
    .to("kafka:patient-admissions")
  .when(header("messageType").isEqualTo("ORU"))
    .to("kafka:lab-results")
.end();

// Пример оптимизации производительности в Camel
from("direct:start")
.threads(10) // Параллельная обработка
.split(body().tokenize("\n")) // Разделение на части для параллельной обработки
.streaming() // Потоковая обработка для больших сообщений
.to("direct:process");

Партиции -> Пропускная способность -> Потребители
    1     |           10K            |      1
    5     |           50K            |      5
   10     |          100K            |     10

log.retention.hours=168  # Хранение сообщений в течение 7 дней
log.retention.bytes=1073741824  # Ограничение по размеру (1 ГБ)

// Пример обеспечения надежности в Camel
from("jms:queue:orders?acknowledgementMode=CLIENT_ACKNOWLEDGE")
.transacted()
.to("processing-service")
.to("jms:queue:processed-orders");

Сравнение пропускной способности:
Camel: 500-5000 сообщений/сек на узел (зависит от сложности обработки)
Kafka: 100K-1M+ сообщений/сек на брокер (зависит от размера сообщений)

// Настройка обработки отказов в потребителе Kafka
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // При потере позиции начинать с самого начала
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Ручное подтверждение обработки для избежания потери сообщений

from("jms:incoming")
  .errorHandler(deadLetterChannel("jms:error")
      .useOriginalMessage()
      .maximumRedeliveries(3)
      .redeliveryDelay(1000)
      .backOffMultiplier(2)
      .useExponentialBackOff())
  .process(myProcessor)
  .to("jms:outgoing");

kafka_consumergroup_lag{group="order-processing"} > 1000

from("direct:orders")
.to("metrics:timer:orderProcessing?action=start")
.process(orderProcessor)
.to("metrics:timer:orderProcessing?action=stop")
.to("kafka:processed-orders");

События → [Kafka] → [Camel] → Внешние системы