EAI（Enterprise Application Integration）完全ガイド：企業システム統合のすべて

EAI (統合戦略)
├── SOA (アーキテクチャスタイル)
│   └── Microservices (SOA発展形)
├── API Management (API管理レイヤー)
├── Message Broker (非同期統合)
└── iPaaS (クラウド統合プラットフォーム)

システムA ←→ システムB
システムA ←→ システムC
システムB ←→ システムC
システムA ←→ システムD
システムB ←→ システムD
システムC ←→ システムD

         システムA
            |
システムD ── HUB ── システムB
            |
         システムC

システムA ── アダプタ ──┐
                       │
システムB ── アダプタ ──┼── ESB（分散メッセージバス）──┼── アダプタ ── システムE
                       │                              │
システムC ── アダプタ ──┘                              └── アダプタ ── システムF

[モバイル/ウェブ/パートナー]
       │
  Experience API（チャネル別カスタマイズ）
       │
   Process API（ビジネスロジック合成）
       │
   System API（バックエンドシステム抽象化）
       │
[SAP / Salesforce / DB / レガシー]

[クライアント]
    │
API Gateway（同期リクエスト）
    │                    Event Mesh（非同期イベント）
    ├── REST API ───────── Kafka / EventBridge
    └── GraphQL             │
                     ┌──────┼──────┐
                  サービスA  サービスB  サービスC

// Apache Camel - メッセージチャネル定義
from("jms:queue:orders")       // 入力チャネル
    .to("jms:queue:validated"); // 出力チャネル

// Apache Camel - Content-Based Router
from("jms:queue:orders")
    .choice()
        .when(xpath("/order/country = 'JP'"))
            .to("jms:queue:orders-jp")
        .when(xpath("/order/country = 'US'"))
            .to("jms:queue:orders-us")
        .when(xpath("/order/country = 'KR'"))
            .to("jms:queue:orders-kr")
        .otherwise()
            .to("jms:queue:orders-other")
    .end();

// Apache Camel - Message Filter
from("jms:queue:orders")
    .filter(xpath("/order/amount > 100000"))
    .to("jms:queue:high-value-orders");

// Apache Camel - XMLからJSONへの変換
from("jms:queue:orders-xml")
    .unmarshal().jacksonXml(Order.class)
    .marshal().json(JsonLibrary.Jackson)
    .to("jms:queue:orders-json");

// Apache Camel - Splitter
from("jms:queue:batch-orders")
    .split(xpath("/orders/order"))
    .to("jms:queue:single-order");

// Apache Camel - Aggregator
from("jms:queue:single-order-result")
    .aggregate(header("batchId"), new OrderAggregationStrategy())
    .completionSize(10)
    .completionTimeout(5000)
    .to("jms:queue:batch-result");

// Apache Camel - Dead Letter Channel
errorHandler(deadLetterChannel("jms:queue:dead-letters")
    .maximumRedeliveries(3)
    .redeliveryDelay(1000)
    .retryAttemptedLogLevel(LoggingLevel.WARN));

from("jms:queue:orders")
    .process(exchange -> {
        // 処理ロジック - 失敗時は自動的にDLCへ移動
        Order order = exchange.getIn().getBody(Order.class);
        orderService.process(order);
    })
    .to("jms:queue:processed-orders");

// Apache Camel - Idempotent Consumer（メモリベース）
from("jms:queue:orders")
    .idempotentConsumer(
        header("orderId"),
        MemoryIdempotentRepository.memoryIdempotentRepository(200)
    )
    .to("jms:queue:unique-orders");

// Redisベースの冪等性（プロダクション推奨）
from("jms:queue:orders")
    .idempotentConsumer(
        header("orderId"),
        new RedisIdempotentRepository(redisTemplate, "processed-orders")
    )
    .to("jms:queue:unique-orders");

// Apache Camel - Wire Tap
from("jms:queue:orders")
    .wireTap("jms:queue:order-audit")    // 監査用複製
    .to("jms:queue:order-processing");   // 元のフロー継続

CDMなし（N:Nマッピング）：
SAP (IDOC) <-> Salesforce (SOQL)
SAP (IDOC) <-> WMS (CSV)
Salesforce (SOQL) <-> WMS (CSV)
= 3マッピング（3システム基準）

CDM適用（N:1マッピング）：
SAP (IDOC) -> CDM (JSON) -> Salesforce
WMS (CSV) -> CDM (JSON) -> Salesforce
SAP (IDOC) -> CDM (JSON) -> WMS
= 各システムからCDMへの1マッピングのみ実装

{
  "canonicalOrder": {
    "orderId": "ORD-2026-001",
    "orderDate": "2026-03-23T10:30:00Z",
    "customer": {
      "customerId": "CUST-001",
      "name": "田中太郎",
      "email": "tanaka@example.com"
    },
    "items": [
      {
        "productId": "PROD-100",
        "productName": "ノートパソコン",
        "quantity": 2,
        "unitPrice": 180000,
        "currency": "JPY"
      }
    ],
    "totalAmount": 360000,
    "status": "CREATED"
  }
}

<!-- SAP IDOCをCDMに変換するXSLT -->
<xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
  <xsl:template match="/ORDERS05">
    <canonicalOrder>
      <orderId><xsl:value-of select="IDOC/E1EDK01/BELNR"/></orderId>
      <orderDate><xsl:value-of select="IDOC/E1EDK01/DATUM"/></orderDate>
      <customer>
        <customerId><xsl:value-of select="IDOC/E1EDKA1/PARVW"/></customerId>
        <name><xsl:value-of select="IDOC/E1EDKA1/NAME1"/></name>
      </customer>
    </canonicalOrder>
  </xsl:template>
</xsl:stylesheet>

%dw 2.0
output application/json
---
{
  canonicalOrder: {
    orderId: payload.ORDERS05.IDOC.E1EDK01.BELNR,
    orderDate: payload.ORDERS05.IDOC.E1EDK01.DATUM as Date,
    customer: {
      customerId: payload.ORDERS05.IDOC.E1EDKA1.PARVW,
      name: payload.ORDERS05.IDOC.E1EDKA1.NAME1
    },
    items: payload.ORDERS05.IDOC.E1EDP01 map ((item) -> {
      productId: item.MATNR,
      quantity: item.MENGE as Number,
      unitPrice: item.NETPR as Number
    })
  }
}

{
  "canonicalOrder": {
    "orderId": sapOrder.BELNR,
    "orderDate": sapOrder.DATUM,
    "customer": {
      "customerId": sapOrder.PARVW,
      "name": sapOrder.NAME1
    },
    "totalAmount": $sum(sapOrder.items.(MENGE * NETPR))
  }
}

# Debeziumコネクタ設定（MySQL）
name: mysql-connector
config:
  connector.class: io.debezium.connector.mysql.MySqlConnector
  database.hostname: mysql-server
  database.port: 3306
  database.user: debezium
  database.password: secret
  database.server.id: 1
  topic.prefix: myapp
  database.include.list: ecommerce
  table.include.list: ecommerce.orders,ecommerce.customers
  schema.history.internal.kafka.bootstrap.servers: kafka:9092
  schema.history.internal.kafka.topic: schema-changes

[MySQL DB] -> [Debezium Connector] -> [Kafka Topic] -> [Consumer App]
                                                      -> [Elasticsearch]
                                                      -> [Data Warehouse]

// Spring Boot + Apache Camelの例
@Component
public class OrderIntegrationRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {
        // エラーハンドリング
        errorHandler(deadLetterChannel("kafka:dead-letters")
            .maximumRedeliveries(3)
            .redeliveryDelay(2000));

        // SAPから注文受信 -> 変換 -> CRMへ送信
        from("sap-idoc:ORDERS05")
            .routeId("sap-to-crm-orders")
            .log("SAP注文受信: orderId=${header.orderId}")
            .unmarshal().jacksonXml(SapOrder.class)
            .bean(OrderTransformer.class, "toCanonical")
            .marshal().json(JsonLibrary.Jackson)
            .choice()
                .when(simple("${body[totalAmount]} > 10000000"))
                    .to("kafka:high-value-orders")
                .otherwise()
                    .to("kafka:standard-orders")
            .end()
            .to("salesforce:upsert:Order__c?sObjectIdName=SAP_Order_Id__c");
    }
}

[ウェブサイト] -> REST API -> [API Gateway]
                                 │
                          [Apache Camel Hub]
                          ┌──────┼──────────┐
                          │      │          │
                       [Kafka]  [Kafka]   [Kafka]
                       orders   inventory  shipping
                          │      │          │
                     [SAP ERP] [WMS]  [Salesforce CRM]
                          │      │          │
                          └──────┼──────────┘
                                 │
                          [Monitoring]
                       Prometheus + Grafana

@Component
public class EcommerceIntegrationRoutes extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        // ===== グローバルエラーハンドリング =====
        errorHandler(deadLetterChannel("kafka:dead-letter-queue")
            .maximumRedeliveries(3)
            .redeliveryDelay(2000)
            .backOffMultiplier(2)
            .useExponentialBackOff()
            .retryAttemptedLogLevel(LoggingLevel.WARN)
            .logRetryStackTrace(true));

        // ===== 第1段階：注文受信 -> SAP ERP登録 =====
        from("rest:post:/api/orders")
            .routeId("order-intake")
            .log("新規注文受信: ${body}")
            .unmarshal().json(JsonLibrary.Jackson, OrderRequest.class)
            .bean(OrderValidator.class, "validate")
            .bean(OrderTransformer.class, "toCanonical")
            .marshal().json(JsonLibrary.Jackson)
            .to("kafka:orders?brokers=localhost:9092")
            .setBody(simple("注文が受け付けられました。注文番号: ${header.orderId}"));

        // SAP ERP注文登録
        from("kafka:orders?brokers=localhost:9092&groupId=sap-consumer")
            .routeId("sap-order-registration")
            .idempotentConsumer(
                header("orderId"),
                new RedisIdempotentRepository(redisTemplate, "sap-orders")
            )
            .unmarshal().json(JsonLibrary.Jackson, CanonicalOrder.class)
            .bean(SapOrderTransformer.class, "toSapIdoc")
            .to("sap-idoc:ORDERS05")
            .log("SAP注文登録完了: ${header.orderId}")
            .to("kafka:inventory-check?brokers=localhost:9092");

        // ===== 第2段階：在庫確認 -> WMS出荷指示 =====
        from("kafka:inventory-check?brokers=localhost:9092&groupId=wms-consumer")
            .routeId("inventory-check-and-ship")
            .unmarshal().json(JsonLibrary.Jackson, CanonicalOrder.class)
            .bean(InventoryService.class, "checkAvailability")
            .choice()
                .when(simple("${header.inventoryAvailable} == true"))
                    .log("在庫確保: ${header.orderId}")
                    .bean(WmsTransformer.class, "toShipmentRequest")
                    .to("http://wms-api.internal:8080/api/shipments")
                    .to("kafka:shipping-started?brokers=localhost:9092")
                .otherwise()
                    .log("在庫不足: ${header.orderId}")
                    .to("kafka:backorder-queue?brokers=localhost:9092")
            .end();

        // ===== 第3段階：配送開始 -> Salesforceステータス更新 =====
        from("kafka:shipping-started?brokers=localhost:9092&groupId=crm-consumer")
            .routeId("crm-status-update")
            .unmarshal().json(JsonLibrary.Jackson, ShipmentEvent.class)
            .bean(SalesforceTransformer.class, "toOpportunityUpdate")
            .to("salesforce:upsert:Opportunity?sObjectIdName=Order_Id__c")
            .log("Salesforceステータス更新完了: ${header.orderId}");

        // ===== サーキットブレーカー =====
        from("kafka:critical-orders?brokers=localhost:9092")
            .routeId("circuit-breaker-route")
            .circuitBreaker()
                .resilience4jConfiguration()
                    .failureRateThreshold(50)
                    .waitDurationInOpenState(30000)
                    .slidingWindowSize(10)
                .end()
                .to("http://payment-service:8080/api/process")
            .onFallback()
                .log("決済サービス障害 - フォールバック処理")
                .to("kafka:payment-retry-queue?brokers=localhost:9092")
            .end();
    }
}

# docker-compose-monitoring.yml
version: '3.8'
services:
  prometheus:
    image: prom/prometheus:latest
    ports:
      - '9090:9090'
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

  # Camelメトリクス公開
  camel-app:
    build: .
    ports:
      - '8080:8080'
      - '8081:8081' # Actuator/Prometheusメトリクス
    environment:
      - MANAGEMENT_ENDPOINTS_WEB_EXPOSURE_INCLUDE=health,prometheus
      - MANAGEMENT_METRICS_EXPORT_PROMETHEUS_ENABLED=true

[マイクロサービスA] ──── API Gateway（同期）──── [マイクロサービスB]
        │                                              │
        └──── Event Mesh (Kafka/EventBridge) ──────────┘
                          │
                   [レガシーERP]  <- Apache Camelアダプタ

[外部システム] -> API Gateway -> Lambda（変換）-> EventBridge -> Step Functions
                                                                     │
                                                          ┌──────────┼──────────┐
                                                       Lambda A   Lambda B   Lambda C
                                                       (SAP)     (SF)       (WMS)
                                                          │         │         │
                                                          └─────────┼─────────┘
                                                                    │
                                                              DynamoDB（状態）
                                                                    │
                                                          CloudWatch（モニタリング）

[AIエージェント (Claude, GPT)]
        │
    MCP Protocol
        │
[MCPサーバー：EAIオーケストレーター]
    ├── Tool: createIntegrationFlow
    ├── Tool: monitorPipeline
    ├── Tool: triggerDataSync
    └── Tool: debugFailedMessages
        │
[Apache Camel / MuleSoft / AWS Step Functions]

[インターネットバンキング] ──┐
[モバイルバンキング]     ───┤── MCI ── EAI ── コアバンキング
[ATM]                    ───┤                   │
[テラーシステム]          ───┘              ├── CRM
                                           ├── リスク
                                           └── FEP -> KFTC/カード会社/保険会社