🔍 什么是 Apache Camel？

Apache Camel 是一个基于 企业集成模式（EIPs） 的开源集成框架。它提供了一种标准化且易于阅读的方式，通过 路由 来连接不同的系统，路由有点像是数据的 “管道”。

我们可以在哪些场景使用 Camel？

• • 轻松与 Spring Boot 集成
• • 使用 Java 领域特定语言（DSL）实现声明式路由逻辑
• • 300 多个内置组件（如 Kafka、HTTP、文件等）
• • 简单的测试和调试
• • 非常适合微服务和数据管道

你应该了解的核心概念

在开始编写代码之前，以下是 Apache Camel 中的一些关键概念：

• • 路由：定义数据来源和去向的管道。
• • 端点：路由开始或结束的统一资源标识符（URI）。
• • 组件：像 file、kafka、http 等这样的连接器。
• • 交换对象：消息的容器。
• • 处理器：处理消息的 Java 函数。

示例路由：

from("file:input")
    .process(exchange -> {
        String body = exchange.getIn().getBody(String.class);
        // 修改或转换消息
    })
    .to("log:output");

项目设置（Maven）

在你的 pom.xml 中添加以下依赖项：

<dependency>
    <groupId>org.apache.camel.springboot</groupId>
    <artifactId>camel-spring-boot-starter</artifactId>
    <version>4.0.0</version>
</dependency>
<dependency>
    <groupId>org.apache.camel.springboot</groupId>
    <artifactId>camel-file-starter</artifactId>
</dependency>
<!-- 如果需要，添加其他组件，如 camel-kafka-starter 或 camel-http-starter -->

你的第一个 Camel 路由

以下是如何创建一个简单的 Camel 路由，该路由每 5 秒记录一条消息：

@Component
public class TimerRoute extends RouteBuilder {
    @Override
    public void configure() {
        from("timer:hello?period=5000")
            .log("Hello from Camel! The time is ${date:now}");
    }
}

常用的 Camel 组件

以下是一些广泛使用的 Camel 组件：

Camel 中的错误处理

Camel 提供了强大的错误处理策略。以下是一个简单的示例：

@Override
public void configure() {
    onException(Exception.class)
        .log("Something went wrong: ${exception.message}")
        .handled(true);
    from("file:input")
        .to("bean:myProcessor")
        .to("file:output");
}

你还可以定义重试次数、死信队列等。

测试 Camel 路由

你可以使用 @CamelSpringBootTest 轻松地为你的路由编写单元测试。

@SpringBootTest
@CamelSpringBootTest
public class MyRouteTest {
    @Autowired
    private ProducerTemplate template;
    @Test
    public void testSimpleRoute() {
        String result = template.requestBody("direct:start", "Hello", String.class);
        assertEquals("Processed: Hello", result);
    }
}

📬 实际应用案例：

1. 使用 Apache Camel 通过 Kafka 发送设备通知

在这个示例中，我们将使用 Apache Camel 和 Spring Boot 构建一个可靠的 Kafka 消费者。用例很简单但很强大：想象一个系统，其中各种微服务将设备通知消息发布到 Kafka 主题。每条消息都包含设备 ID 和要通过推送通知发送的消息内容。

Apache Camel 将：

• • 从 Kafka 主题消费消息。
• • 反序列化并丰富消息内容。
• • 处理业务逻辑（发送通知）。
• • 失败时重试。
• • 记录关联 ID 以实现可追溯性。

让我们详细分析一下。🚀

带有重试和日志记录的 Kafka 消费者路由

@Component
@RequiredArgsConstructor
public class DeviceNotificationKafkaRoute extends RouteBuilder {
    private final MDCProcessor mdcProcessor;
    private final ExchangeParser exchangeParser;
    private final CamelRetryProcessor camelRetryProcessor;
    private final DeviceNotificationProcessor deviceNotificationProcessor;
    @Override
    public void configure() {
      configureExceptionHandling();
      configureKafkaRoute();
    }
    private void configureExceptionHandling() {
      onException(Exception.class)
          .maximumRedeliveries(3)
          .redeliveryDelay(2000)
          .log("Retrying due to: ${exception.message}")
          .process(camelRetryProcessor)
          .handled(true);
    }
    private void configureKafkaRoute() {
      from("kafka:device-notification?brokers=localhost:9092&groupId=device-group&concurrentConsumers=20&autoOffsetReset=earliest")
          .routeId("DeviceNotificationKafkaRoute")
          .process(this::setupProperties)
          .process(mdcProcessor)
          .process(exchangeParser)
          .log("Received device notification payload: ${body}")
          .process(deviceNotificationProcessor)
          .end();
    }
    private void setupProperties(Exchange exchange) {
      exchange.setProperty(Constants.TARGET_CLASS, DeviceNotificationPayload.class);
      exchange.setProperty(Constants.CAMEL_RETRY_PROCESSOR_QUEUE_NAME, "device-notification");
      exchange.setProperty(Constants.CAMEL_RETRY_PROCESSOR_MAX_RETRY_COUNT, 3);
    }
}

业务逻辑处理器

@Component
@RequiredArgsConstructor
public class DeviceNotificationProcessor implements Processor {
  private final NotificationService notificationService;
    @Override
    public void process(Exchange exchange) {
      DeviceNotificationPayload payload = exchange.getIn().getBody(DeviceNotificationPayload.class);
      // 模拟推送通知逻辑
      notificationService.sendPushNotification(payload.getDeviceId(), payload.getMessage());
    }
}

这里发生了什么？

• • Kafka 源：我们使用 kafka: 组件监听 device-notification 主题。
• • 重试机制：如果发生失败，Camel 将重试 3 次，每次重试间隔 2 秒。
• • 异常处理：捕获异常并记录日志。达到最大重试次数后，错误将被视为已处理。
• • 数据丰富：MDCProcessor 添加关联 ID 以便进行日志追踪。
• • 反序列化：ExchangeParser 将原始的 Kafka JSON 字符串转换为 DeviceNotificationPayload 对象。
• • 业务逻辑：DeviceNotificationProcessor 将消息传递给推送通知服务。
• • 并发消费者：concurrentConsumers=20 用于提高吞吐量和并行处理能力。可以通过配置并发消费者，允许多个线程同时从 Kafka 主题消费消息。

为什么这种模式有效

这种方法具有以下优点：

• • 解耦：每个组件都有单一职责。
• • 弹性：通过重试处理瞬态故障。
• • 可观察性：日志包含关联 ID，便于追踪。
• • 简洁：Apache Camel 使流程易于理解。

你可以通过以下方式扩展此功能：

• • 为失败消息使用死信队列（DLQ）。
• • 通过 Prometheus + Grafana 进行告警。
• • 使用 Micrometer 收集高级指标。
• • 根据设备类型进行动态路由。

2. Apache Camel WebSocket 集成示例

使用 Apache Camel，我们可以连接到 WebSocket 服务器并处理传入的数据。在以下示例中，Camel 连接到一个 WebSocket 端点，记录传入的消息，然后将它们传递给一个处理器进行进一步处理。

WebSocket 路由（消费者）

@Component
public class WebSocketRoute extends RouteBuilder {
    @Override
    public void configure() {
      from("websocket://0.0.0.0:9292/ws/notifications?sendToAll=true")
          .routeId("WebSocketNotificationRoute")
          .log("📩 Received WebSocket Message: ${body}")
          .process(exchange -> {
              String message = exchange.getIn().getBody(String.class);
              // 模拟处理
              System.out.println("💡 Processing WebSocket message: " + message);
          });
    }
}

你可以通过使用 WebSocket 客户端连接到 ws://localhost:9292/ws/notifications 来测试此路由。

3. Apache Camel 与 Quartz 调度器示例

Apache Camel 可以与 Quartz 无缝集成，以便使用 cron 表达式或时间间隔来调度任务。在这个示例中，Camel 每分钟触发一次路由并执行一些逻辑。

基于 Quartz 的定时路由

@Component
public class QuartzSchedulerRoute extends RouteBuilder {
    @Override
    public void configure() {
      from("quartz://notificationScheduler?cron=0+0/1+*+*+*+?")
          .routeId("ScheduledNotificationRoute")
          .log("⏰ Quartz Trigger Activated at ${date:now:yyyy-MM-dd HH:mm:ss}")
          .process(exchange -> {
              // 模拟定时任务
              System.out.println("🚀 Running scheduled notification task...");
          });
    }
}

此配置会 每分钟触发一次 该路由。你可以根据需要调整 cron 表达式（例如，0/10 * * * * ?_ 表示每 10 秒触发一次）。_

4. 由 REST 端点（HTTP）触发的 Apache Camel 路由

在这个示例中，Camel 路由由对 /api/trigger-notification 端点的 HTTP POST 请求触发。传入的负载首先被反序列化，然后由一个处理器处理业务逻辑。

路由定义

@Component
public class RestNotificationTriggerRoute extends RouteBuilder {
    @Override
    public void configure() {
      restConfiguration()
          .component("servlet")
          .contextPath("/api")
          .port(8080);
      rest("/trigger-notification")
          .post()
          .type(TriggerNotificationRequest.class)
          .route()
          .routeId("RestNotificationTriggerRoute")
          .log("📨 Received HTTP trigger: ${body}")
          .process(new NotificationTriggerProcessor())
          .endRest();
    }
}

处理器类

public class NotificationTriggerProcessor implements Processor {
  @Override
  public void process(Exchange exchange) {
    TriggerNotificationRequest request = exchange.getIn().getBody(TriggerNotificationRequest.class);
    
    // 模拟通知逻辑
    System.out.println("🔔 Sending notification to user: " + request.getUserId());
    System.out.println("📝 Message: " + request.getMessage());
  }
}

如何测试

你可以使用以下简单的 curl 命令进行测试：

curl -X POST http://localhost:8080/api/trigger-notification \
  -H "Content-Type: application/json" \
  -d '{"userId": "123", "message": "Hello from Camel!"}'

5. 在 Apache Camel 中使用断路器

Apache Camel 原生支持 断路器 模式。这使你能够保护系统免受故障影响，优雅地处理停机情况，并确保即使外部系统出现故障，你的服务仍然保持响应。借助 Camel 内置的断路器，你可以定义回退行为、管理重试次数，并配置阈值以防止级联故障。

@Component
public class CircuitBreakerRoute extends RouteBuilder {
    @Override
    public void configure() {
      from("direct:start")
          .circuitBreaker()
            .maximumFailures(3)
            .delay(5000)  // 5 秒后重试
            .onFailureOnly()
            .log("Circuit breaker activated due to failure.")
          .end()
          .to("http://external-service/endpoint")
          .log("Response from external service: ${body}");
    }
}

在这个示例中，如果外部服务失败 3 次，断路器将被触发，并且在重试之前路由将等待 5 秒。onFailureOnly() 确保记录失败情况，并且在断路器关闭之前不会再发送进一步的请求。

总结

Apache Camel 是 Spring Boot 生态系统中最被低估的工具之一。如果你正在构建以集成为主的应用程序，使用 Camel 可以极大地简化你的逻辑，并使你的代码更易于维护。

“如果说 Spring Boot 是引擎，那么 Apache Camel 就是高速公路。”