Apache Camel中实现动态路由、多目标消息发送与精细化重试策略

// 假设 CustomerConfigRetrieverBean 接收 RemappedMessage 作为输入
public class CustomerConfigRetrieverBean {
    public List<ImmutablePair<CustomerConfig, RemappedMessage>> retrieveAndCombine(RemappedMessage remappedMessage) {
        // 根据 remappedMessage 中的 agentId 获取所有 CustomerConfig
        List<CustomerConfig> customerConfigs = retrieveCustomerConfigsByAgentId(remappedMessage.getAgentId());

        List<ImmutablePair<CustomerConfig, RemappedMessage>> combinedList = new ArrayList<>();
        for (CustomerConfig config : customerConfigs) {
            // 为每个客户配置创建一个 ImmutablePair，将其与原始的 remappedMessage 关联
            combinedList.add(ImmutablePair.of(config, remappedMessage));
        }
        return combinedList; // 这个列表将作为 Splitter 的输入
    }

    // 模拟方法，实际应从配置源获取
    private List<CustomerConfig> retrieveCustomerConfigsByAgentId(String agentId) {
        // ... 实现配置获取逻辑 ...
        return Arrays.asList(
            new CustomerConfig("customer1", "http://api.customer1.com/data", "oauth_url_1", "client_id_1", "client_secret_1", "fieldA,fieldB"),
            new CustomerConfig("customer2", "http://api.customer2.com/data", "oauth_url_2", "client_id_2", "client_secret_2", "fieldC")
        );
    }
}

// 假设消息体现在是一个 ImmutablePair<CustomerConfig, RemappedMessage>
.setHeader("Authorization", simple("Bearer ${body.left.oauthToken}")) // 假设 CustomerConfig 中有获取到的OAuth Token
// 或者对于 Basic Auth:
// .setHeader("Authorization", simple("Basic ${body.left.base64EncodedCredentials}"))

// 设置目标URI
.setHeader(Exchange.HTTP_URI, simple("${body.left.targetUrl}")) // 使用 CustomerConfig 中的 targetUrl

// 假设 CustomerConfig 包含 OAuth 相关的URL和凭据
public class OAuthTokenRetrieverBean {
    public void retrieveToken(@Body ImmutablePair<CustomerConfig, RemappedMessage> pair, @Headers Map<String, Object> headers) {
        CustomerConfig config = pair.getLeft();
        // 实际的OAuth请求逻辑
        String oauthToken = performOAuthRequest(config.getOauthUrl(), config.getClientId(), config.getClientSecret());
        headers.put("Authorization", "Bearer " + oauthToken);
    }

    private String performOAuthRequest(String oauthUrl, String clientId, String clientSecret) {
        // ... 实现 OAuth 客户端逻辑，例如使用 HttpClient ...
        return "mock_oauth_token"; // 模拟返回
    }
}

from("activemq:queue:" + appConfig.getQueueName())
    .bean(IncomingMessageConverter.class) // 转换为 RemappedMessage
    .bean(UserIdValidator.class) // 验证用户ID
    .bean(CustomerConfigRetrieverBean.class) // 返回 List<ImmutablePair<CustomerConfig, RemappedMessage>>
    .split(body()) // 拆分列表，每个元素成为一个新的Exchange
        .parallelProcessing() // 可选：并行处理每个客户的发送任务
        .errorHandler(deadLetterChannel("activemq:queue:dlq").maximumRedeliveries(3).redeliveryDelay(2000)) // 为每个拆分消息配置重试
        .bean(EndpointFieldsTailor.class) // 根据 CustomerConfig 裁剪消息字段
        .bean(OAuthTokenRetrieverBean.class) // 获取OAuth Token并设置Authorization头
        .filter(simple("${body.left.filterCriteria.matches(${body.right.messageContent})}")) // 根据客户条件过滤消息
            .setHeader(Exchange.HTTP_URI, simple("${body.left.targetUrl}")) // 设置目标URL
            .toD("http://dummyhost?throwExceptionOnFailure=false") // 使用 toD 动态发送，并允许失败继续
        .endFilter()
    .end() // 结束 Splitter EIP
    .log("所有客户消息处理完毕");

import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.commons.lang3.tuple.ImmutablePair;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

// 假设的配置类和消息类
class AppConfig {
    String getQueueName() { return "myQueue"; }
}

class RemappedMessage {
    String agentId;
    String messageContent;
    // ... 其他字段
    public RemappedMessage(String agentId, String messageContent) {
        this.agentId = agentId;
        this.messageContent = messageContent;
    }
    public String getAgentId() { return agentId; }
    public String getMessageContent() { return messageContent; }
}

class CustomerConfig {
    String customerId;
    String targetUrl;
    String oauthUrl;
    String clientId;
    String clientSecret;
    String requiredFields; // 逗号分隔的字段列表
    String filterCriteria; // 过滤条件，例如一个正则表达式或简单表达式

    public CustomerConfig(String customerId, String targetUrl, String oauthUrl, String clientId, String clientSecret, String requiredFields) {
        this.customerId = customerId;
        this.targetUrl = targetUrl;
        this.oauthUrl = oauthUrl;
        this.clientId = clientId;
        this.clientSecret = clientSecret;
        this.requiredFields = requiredFields;
        this.filterCriteria = ".*"; // 默认不过滤
    }
    // Getters
    public String getTargetUrl() { return targetUrl; }
    public String getOauthUrl() { return oauthUrl; }
    public String getClientId() { return clientId; }
    public String getClientSecret() { return clientSecret; }
    public String getRequiredFields() { return requiredFields; }
    public String getFilterCriteria() { return filterCriteria; }
    public void setFilterCriteria(String filterCriteria) { this.filterCriteria = filterCriteria; }
}

// 模拟的Bean
class IncomingMessageConverter {
    public RemappedMessage convert(String rawMessage) {
        System.out.println("Converting incoming message: " + rawMessage);
        return new RemappedMessage("agentX", "original_content_with_fieldA_fieldB_fieldC");
    }
}

class UserIdValidator {
    public void validate(RemappedMessage message) {
        System.out.println("Validating user ID for agent: " + message.getAgentId());
        // 模拟验证逻辑，如果失败可以抛出异常
    }
}

// CustomerConfigRetrieverBean 见上文

class EndpointFieldsTailor {
    public ImmutablePair<CustomerConfig, RemappedMessage> tailor(ImmutablePair<CustomerConfig, RemappedMessage> pair) {
        CustomerConfig config = pair.getLeft();
        RemappedMessage message = pair.getRight();
        System.out.println("Tailoring fields for customer " + config.customerId + ". Required: " + config.getRequiredFields());
        // 根据 config.getRequiredFields() 修改 message 的内容
        // 这里只是模拟，实际可能涉及 JSON/XML 解析和字段操作
        String tailoredContent = "tailored_content_for_" + config.customerId + "_with_" + config.getRequiredFields();
        return ImmutablePair.of(config, new RemappedMessage(message.getAgentId(), tailoredContent));
    }
}

// OAuthTokenRetrieverBean 见上文

public class DynamicRoutingTutorialRoute extends RouteBuilder {

    private final AppConfig appConfig;

    public DynamicRoutingTutorialRoute(AppConfig appConfig) {
        this.appConfig = appConfig;
    }

    @Override
    public void configure() throws Exception {
        // 全局错误处理，捕获未被特定路由块处理的异常
        onException(Exception.class)
            .maximumRedeliveries(0) // 不重试
            .handled(true)
            .log("全局错误捕获: ${exception.message}");

        from("activemq:queue:" + appConfig.getQueueName())
            .routeId("mainProcessingRoute")
            .log("接收到来自队列的消息")
            .bean(IncomingMessageConverter.class)
            .bean(UserIdValidator.class)
            .bean(CustomerConfigRetrieverBean.class) // 返回 List<ImmutablePair<CustomerConfig, RemappedMessage>>
            .split(body()).parallelProcessing() // 拆分列表，并行处理
                .setHeader("customerId", simple("${body.left.customerId}")) // 将客户ID提升到Header，便于日志和跟踪
                .log("开始处理客户: ${header.customerId}")
                // 为每个拆分消息配置局部重试
                .errorHandler(deadLetterChannel("activemq:queue:customerDlq")
                    .maximumRedeliveries(3)
                    .redeliveryDelay(2000)
                    .logStackTrace(true)
                    .onRedelivery(exchange -> {
                        System.out.println("Retrying send for customer " + exchange.getIn().getHeader("customerId") + " (attempt " + exchange.getProperty(Exchange.REDELIVERY_COUNTER) + ")");
                    }))
                .bean(EndpointFieldsTailor.class) // 根据 CustomerConfig 裁剪消息字段
                .bean(OAuthTokenRetrieverBean.class) // 获取OAuth Token并设置Authorization头
                .filter(simple("${body.right.messageContent} contains '${body.left.filterCriteria}'")) // 根据客户条件过滤消息
                    .log("客户 ${header.customerId} 满足过滤条件，准备发送")
                    .setHeader(Exchange.HTTP_URI, simple("${body.left.targetUrl}")) // 设置目标URL
                    .setHeader(Exchange.HTTP_METHOD, constant("POST")) // 设置HTTP方法
                    .setBody(simple("${body.right.messageContent}")) // 将 RemappedMessage 的内容作为HTTP Body
                    .toD("http://dummyhost?throwExceptionOnFailure=true") // 使用 toD 动态发送，失败则抛出异常触发重试
                    .log("成功发送消息给客户: ${header.customerId}, 响应: ${body}")
                .endFilter()
                .log("客户 ${header.customerId} 处理完成 (可能已过滤或发送)")
            .end() // 结束 Splitter EIP
            .log("所有客户消息处理完毕，主路由结束");
    }
}