// 同步服务调用示例
@Service
public class SynchronousIntegrationService {
    
    @Autowired
    private RestTemplate restTemplate;
    
    @Autowired
    private IntegrationConfig config;
    
    // 同步调用外部系统API
    public CustomerInfo getCustomerInfo(String customerId) {
        try {
            String url = config.getCustomerServiceUrl() + "/customers/" + customerId;
            
            // 设置请求头
            HttpHeaders headers = new HttpHeaders();
            headers.set("Authorization", "Bearer " + getAccessToken());
            headers.set("Content-Type", "application/json");
            
            HttpEntity<String> entity = new HttpEntity<>(headers);
            
            // 执行同步调用
            ResponseEntity<CustomerInfo> response = restTemplate.exchange(
                url, HttpMethod.GET, entity, CustomerInfo.class);
            
            // 记录调用日志
            logIntegrationCall("GET_CUSTOMER_INFO", customerId, 
                response.getStatusCode().value(), System.currentTimeMillis());
            
            return response.getBody();
        } catch (Exception e) {
            // 记录错误日志
            logIntegrationError("GET_CUSTOMER_INFO", customerId, e);
            throw new IntegrationException("获取客户信息失败", e);
        }
    }
    
    // 同步更新客户信息
    public boolean updateCustomerInfo(String customerId, CustomerInfo customerInfo) {
        try {
            String url = config.getCustomerServiceUrl() + "/customers/" + customerId;
            
            HttpHeaders headers = new HttpHeaders();
            headers.set("Authorization", "Bearer " + getAccessToken());
            headers.set("Content-Type", "application/json");
            
            HttpEntity<CustomerInfo> entity = new HttpEntity<>(customerInfo, headers);
            
            ResponseEntity<Void> response = restTemplate.exchange(
                url, HttpMethod.PUT, entity, Void.class);
            
            logIntegrationCall("UPDATE_CUSTOMER_INFO", customerId, 
                response.getStatusCode().value(), System.currentTimeMillis());
            
            return response.getStatusCode().is2xxSuccessful();
        } catch (Exception e) {
            logIntegrationError("UPDATE_CUSTOMER_INFO", customerId, e);
            throw new IntegrationException("更新客户信息失败", e);
        }
    }
}

异步集成

// 异步服务调用示例
@Service
public class AsynchronousIntegrationService {
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    @Autowired
    private IntegrationEventRepository eventRepository;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    // 发送异步事件
    @Async
    public CompletableFuture<Boolean> sendOrderEvent(OrderEvent orderEvent) {
        try {
            // 创建事件记录
            IntegrationEvent event = new IntegrationEvent();
            event.setEventId(UUID.randomUUID().toString());
            event.setEventType("ORDER_EVENT");
            event.setPayload(objectMapper.writeValueAsString(orderEvent));
            event.setStatus(EventStatus.PENDING);
            event.setCreateTime(new Date());
            
            // 保存事件记录
            eventRepository.save(event);
            
            // 发送到消息队列
            kafkaTemplate.send("order-events", event.getEventId(), orderEvent);
            
            // 更新事件状态
            event.setStatus(EventStatus.SENT);
            event.setSendTime(new Date());
            eventRepository.save(event);
            
            log.info("订单事件发送成功: {}", orderEvent.getOrderId());
            return CompletableFuture.completedFuture(true);
        } catch (Exception e) {
            log.error("订单事件发送失败: {}", orderEvent.getOrderId(), e);
            
            // 更新事件状态为失败
            IntegrationEvent event = eventRepository.findByEventId(orderEvent.getEventId());
            if (event != null) {
                event.setStatus(EventStatus.FAILED);
                event.setErrorMessage(e.getMessage());
                eventRepository.save(event);
            }
            
            return CompletableFuture.completedFuture(false);
        }
    }
    
    // 处理回调事件
    @KafkaListener(topics = "order-callbacks")
    public void handleOrderCallback(ConsumerRecord<String, String> record) {
        try {
            // 解析回调数据
            OrderCallback callback = objectMapper.readValue(record.value(), OrderCallback.class);
            
            // 处理回调逻辑
            processOrderCallback(callback);
            
            log.info("订单回调处理完成: {}", callback.getOrderId());
        } catch (Exception e) {
            log.error("订单回调处理失败: {}", record.key(), e);
            // 可以发送到死信队列进行后续处理
            handleCallbackFailure(record, e);
        }
    }
    
    // 处理订单回调
    private void processOrderCallback(OrderCallback callback) {
        // 根据回调类型执行不同操作
        switch (callback.getCallbackType()) {
            case "ORDER_CONFIRMED":
                confirmOrder(callback.getOrderId());
                break;
            case "ORDER_CANCELLED":
                cancelOrder(callback.getOrderId());
                break;
            case "PAYMENT_COMPLETED":
                updatePaymentStatus(callback.getOrderId(), PaymentStatus.COMPLETED);
                break;
            default:
                log.warn("未知的回调类型: {}", callback.getCallbackType());
        }
    }
}

连接器架构

现代BPM平台通常采用连接器架构来实现灵活的系统集成：

// 通用连接器接口
public interface SystemConnector {
    
    // 连接器初始化
    void initialize(ConnectorConfig config) throws ConnectorException;
    
    // 执行操作
    <T> T execute(String operation, Map<String, Object> parameters, Class<T> responseType) 
        throws ConnectorException;
    
    // 异步执行操作
    <T> CompletableFuture<T> executeAsync(String operation, Map<String, Object> parameters, 
        Class<T> responseType) throws ConnectorException;
    
    // 测试连接
    boolean testConnection() throws ConnectorException;
    
    // 关闭连接
    void close() throws ConnectorException;
}

// REST连接器实现
@Component
public class RestConnector implements SystemConnector {
    
    private RestTemplate restTemplate;
    private ConnectorConfig config;
    private HttpHeaders defaultHeaders;
    
    @Override
    public void initialize(ConnectorConfig config) throws ConnectorException {
        this.config = config;
        
        // 初始化RestTemplate
        this.restTemplate = new RestTemplate();
        
        // 配置请求头
        this.defaultHeaders = new HttpHeaders();
        this.defaultHeaders.set("Content-Type", "application/json");
        
        // 配置认证信息
        if (config.getAuthType() == AuthType.BASIC) {
            String auth = config.getUsername() + ":" + config.getPassword();
            String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes());
            this.defaultHeaders.set("Authorization", "Basic " + encodedAuth);
        } else if (config.getAuthType() == AuthType.BEARER) {
            this.defaultHeaders.set("Authorization", "Bearer " + config.getAccessToken());
        }
        
        // 配置超时时间
        configureTimeout();
    }
    
    @Override
    public <T> T execute(String operation, Map<String, Object> parameters, Class<T> responseType) 
        throws ConnectorException {
        
        try {
            // 构建请求URL
            String url = buildUrl(operation, parameters);
            
            // 构建请求体
            Object requestBody = buildRequestBody(operation, parameters);
            
            // 构建HttpEntity
            HttpEntity<Object> entity = new HttpEntity<>(requestBody, defaultHeaders);
            
            // 确定HTTP方法
            HttpMethod method = determineHttpMethod(operation);
            
            // 执行请求
            ResponseEntity<T> response = restTemplate.exchange(url, method, entity, responseType);
            
            // 记录调用日志
            logCall(operation, parameters, response.getStatusCode().value());
            
            return response.getBody();
        } catch (Exception e) {
            logError(operation, parameters, e);
            throw new ConnectorException("REST调用失败: " + operation, e);
        }
    }
    
    private String buildUrl(String operation, Map<String, Object> parameters) {
        String baseUrl = config.getBaseUrl();
        String endpoint = config.getEndpoints().get(operation);
        
        if (endpoint == null) {
            throw new ConnectorException("未找到操作对应的端点: " + operation);
        }
        
        String url = baseUrl + endpoint;
        
        // 替换路径参数
        for (Map.Entry<String, Object> entry : parameters.entrySet()) {
            String placeholder = "{" + entry.getKey() + "}";
            if (url.contains(placeholder)) {
                url = url.replace(placeholder, entry.getValue().toString());
            }
        }
        
        return url;
    }
    
    private Object buildRequestBody(String operation, Map<String, Object> parameters) {
        // 根据操作类型构建请求体
        RequestBodyConfig bodyConfig = config.getRequestBodyConfigs().get(operation);
        if (bodyConfig == null) {
            return null;
        }
        
        Map<String, Object> requestBody = new HashMap<>();
        for (String field : bodyConfig.getFields()) {
            if (parameters.containsKey(field)) {
                requestBody.put(field, parameters.get(field));
            }
        }
        
        return requestBody;
    }
    
    private HttpMethod determineHttpMethod(String operation) {
        OperationConfig opConfig = config.getOperationConfigs().get(operation);
        if (opConfig != null && opConfig.getHttpMethod() != null) {
            return opConfig.getHttpMethod();
        }
        return HttpMethod.GET;
    }
    
    private void configureTimeout() {
        if (restTemplate.getRequestFactory() instanceof HttpComponentsClientHttpRequestFactory) {
            HttpComponentsClientHttpRequestFactory factory = 
                (HttpComponentsClientHttpRequestFactory) restTemplate.getRequestFactory();
            factory.setConnectTimeout(config.getConnectionTimeout());
            factory.setReadTimeout(config.getReadTimeout());
        }
    }
}

// 数据库连接器实现
@Component
public class DatabaseConnector implements SystemConnector {
    
    private DataSource dataSource;
    private JdbcTemplate jdbcTemplate;
    private ConnectorConfig config;
    
    @Override
    public void initialize(ConnectorConfig config) throws ConnectorException {
        this.config = config;
        
        // 初始化数据源
        this.dataSource = createDataSource(config);
        this.jdbcTemplate = new JdbcTemplate(dataSource);
        
        // 验证连接
        if (!testConnection()) {
            throw new ConnectorException("数据库连接初始化失败");
        }
    }
    
    @Override
    public <T> T execute(String operation, Map<String, Object> parameters, Class<T> responseType) 
        throws ConnectorException {
        
        try {
            SqlOperationConfig sqlConfig = config.getSqlOperations().get(operation);
            if (sqlConfig == null) {
                throw new ConnectorException("未找到SQL操作配置: " + operation);
            }
            
            String sql = sqlConfig.getSql();
            Object[] params = buildParameters(sqlConfig, parameters);
            
            // 执行SQL操作
            if (sqlConfig.getOperationType() == SqlOperationType.SELECT) {
                return jdbcTemplate.queryForObject(sql, params, responseType);
            } else if (sqlConfig.getOperationType() == SqlOperationType.SELECT_LIST) {
                return (T) jdbcTemplate.query(sql, params, new BeanPropertyRowMapper<>(responseType));
            } else if (sqlConfig.getOperationType() == SqlOperationType.UPDATE) {
                int rowsAffected = jdbcTemplate.update(sql, params);
                return (T) Integer.valueOf(rowsAffected);
            } else {
                throw new ConnectorException("不支持的SQL操作类型: " + sqlConfig.getOperationType());
            }
        } catch (Exception e) {
            logError(operation, parameters, e);
            throw new ConnectorException("数据库操作失败: " + operation, e);
        }
    }
    
    private Object[] buildParameters(SqlOperationConfig sqlConfig, Map<String, Object> parameters) {
        List<Object> params = new ArrayList<>();
        for (String paramName : sqlConfig.getParameterNames()) {
            params.add(parameters.get(paramName));
        }
        return params.toArray();
    }
    
    private DataSource createDataSource(ConnectorConfig config) {
        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setJdbcUrl(config.getJdbcUrl());
        dataSource.setUsername(config.getUsername());
        dataSource.setPassword(config.getPassword());
        dataSource.setDriverClassName(config.getDriverClassName());
        dataSource.setMaximumPoolSize(config.getMaxPoolSize());
        dataSource.setMinimumIdle(config.getMinIdle());
        dataSource.setConnectionTimeout(config.getConnectionTimeout());
        return dataSource;
    }
}

协议与数据格式支持

REST/HTTP集成

RESTful API是现代系统集成中最常用的协议之一：

// REST集成服务
@Service
public class RestIntegrationService {
    
    @Autowired
    private RestTemplate restTemplate;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    // 通用REST调用方法
    public <T> ResponseEntity<T> callRestApi(RestApiRequest request, Class<T> responseType) {
        try {
            // 构建URL
            String url = buildUrl(request);
            
            // 构建请求头
            HttpHeaders headers = buildHeaders(request);
            
            // 构建请求体
            HttpEntity<Object> entity = new HttpEntity<>(request.getBody(), headers);
            
            // 执行调用
            return restTemplate.exchange(url, request.getHttpMethod(), entity, responseType);
        } catch (Exception e) {
            throw new IntegrationException("REST API调用失败: " + request.getUrl(), e);
        }
    }
    
    // 支持OAuth2认证的REST调用
    public <T> ResponseEntity<T> callRestApiWithOAuth2(OAuth2RestApiRequest request, 
        Class<T> responseType) {
        
        try {
            // 获取访问令牌
            String accessToken = getOAuth2AccessToken(request.getOAuth2Config());
            
            // 添加认证头
            request.addHeader("Authorization", "Bearer " + accessToken);
            
            // 执行调用
            return callRestApi(request, responseType);
        } catch (Exception e) {
            throw new IntegrationException("OAuth2 REST API调用失败", e);
        }
    }
    
    // 批量REST调用
    public List<RestApiResponse> batchCall(List<RestApiRequest> requests) {
        List<RestApiResponse> responses = new ArrayList<>();
        
        // 并行执行调用
        List<CompletableFuture<RestApiResponse>> futures = requests.stream()
            .map(request -> CompletableFuture.supplyAsync(() -> {
                try {
                    ResponseEntity<String> response = callRestApi(request, String.class);
                    return new RestApiResponse(request.getId(), response.getStatusCodeValue(), 
                        response.getBody(), null);
                } catch (Exception e) {
                    return new RestApiResponse(request.getId(), 500, null, e.getMessage());
                }
            }))
            .collect(Collectors.toList());
        
        // 等待所有调用完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        
        // 收集结果
        futures.forEach(future -> {
            try {
                responses.add(future.get());
            } catch (Exception e) {
                // 处理异常
                log.error("批量调用结果获取失败", e);
            }
        });
        
        return responses;
    }
    
    private String buildUrl(RestApiRequest request) {
        StringBuilder urlBuilder = new StringBuilder(request.getUrl());
        
        // 添加查询参数
        if (request.getQueryParams() != null && !request.getQueryParams().isEmpty()) {
            urlBuilder.append("?");
            String queryParams = request.getQueryParams().entrySet().stream()
                .map(entry -> entry.getKey() + "=" + entry.getValue())
                .collect(Collectors.joining("&"));
            urlBuilder.append(queryParams);
        }
        
        return urlBuilder.toString();
    }
    
    private HttpHeaders buildHeaders(RestApiRequest request) {
        HttpHeaders headers = new HttpHeaders();
        
        // 设置默认内容类型
        headers.setContentType(MediaType.APPLICATION_JSON);
        
        // 添加自定义头信息
        if (request.getHeaders() != null) {
            request.getHeaders().forEach(headers::add);
        }
        
        return headers;
    }
}

SOAP/WebService集成

对于传统的系统集成，SOAP/WebService仍然有其应用场景：

// SOAP集成服务
@Service
public class SoapIntegrationService {
    
    // 动态调用SOAP服务
    public SoapResponse callSoapService(SoapRequest request) throws IntegrationException {
        try {
            // 创建SOAP连接
            SOAPConnectionFactory soapConnectionFactory = SOAPConnectionFactory.newInstance();
            SOAPConnection soapConnection = soapConnectionFactory.createConnection();
            
            // 创建SOAP消息
            MessageFactory messageFactory = MessageFactory.newInstance();
            SOAPMessage soapMessage = messageFactory.createMessage();
            
            // 构建SOAP消息
            buildSoapMessage(soapMessage, request);
            
            // 发送SOAP消息
            SOAPMessage soapResponse = soapConnection.call(soapMessage, request.getEndpoint());
            
            // 解析响应
            SoapResponse response = parseSoapResponse(soapResponse);
            
            // 关闭连接
            soapConnection.close();
            
            return response;
        } catch (Exception e) {
            throw new IntegrationException("SOAP服务调用失败", e);
        }
    }
    
    private void buildSoapMessage(SOAPMessage soapMessage, SoapRequest request) throws Exception {
        SOAPPart soapPart = soapMessage.getSOAPPart();
        SOAPEnvelope envelope = soapPart.getEnvelope();
        SOAPBody soapBody = envelope.getBody();
        
        // 添加命名空间
        if (request.getNamespaces() != null) {
            for (Map.Entry<String, String> entry : request.getNamespaces().entrySet()) {
                envelope.addNamespaceDeclaration(entry.getKey(), entry.getValue());
            }
        }
        
        // 创建请求元素
        Name bodyName = envelope.createName(request.getOperation(), 
            request.getPrefix(), request.getNamespace());
        SOAPBodyElement bodyElement = soapBody.addBodyElement(bodyName);
        
        // 添加参数
        if (request.getParameters() != null) {
            for (Map.Entry<String, Object> entry : request.getParameters().entrySet()) {
                SOAPElement paramElement = bodyElement.addChildElement(entry.getKey());
                paramElement.addTextNode(entry.getValue().toString());
            }
        }
        
        // 保存消息
        soapMessage.saveChanges();
    }
    
    private SoapResponse parseSoapResponse(SOAPMessage soapMessage) throws Exception {
        SoapResponse response = new SoapResponse();
        
        SOAPBody soapBody = soapMessage.getSOAPBody();
        
        // 检查是否有错误
        if (soapBody.hasFault()) {
            SOAPFault fault = soapBody.getFault();
            response.setFault(true);
            response.setFaultCode(fault.getFaultCode());
            response.setFaultString(fault.getFaultString());
        } else {
            // 解析正常响应
            SOAPBodyElement bodyElement = (SOAPBodyElement) soapBody.getChildElements().next();
            response.setPayload(parseElement(bodyElement));
        }
        
        return response;
    }
    
    private Map<String, Object> parseElement(SOAPElement element) {
        Map<String, Object> result = new HashMap<>();
        
        // 获取子元素
        Iterator<?> childElements = element.getChildElements();
        while (childElements.hasNext()) {
            Object child = childElements.next();
            if (child instanceof SOAPElement) {
                SOAPElement childElement = (SOAPElement) child;
                String elementName = childElement.getElementName().getLocalName();
                
                // 如果有子元素，递归解析
                if (childElement.getChildElements().hasNext()) {
                    result.put(elementName, parseElement(childElement));
                } else {
                    // 叶子节点，获取文本值
                    result.put(elementName, childElement.getValue());
                }
            }
        }
        
        return result;
    }
}

数据格式转换与映射

JSON/XML数据处理

在系统集成中，数据格式的转换和映射是关键环节：

// 数据格式转换服务
@Service
public class DataFormatConversionService {
    
    @Autowired
    private ObjectMapper objectMapper;
    
    @Autowired
    private JAXBContext jaxbContext;
    
    // JSON到Java对象转换
    public <T> T jsonToObject(String json, Class<T> clazz) throws ConversionException {
        try {
            return objectMapper.readValue(json, clazz);
        } catch (Exception e) {
            throw new ConversionException("JSON转换失败", e);
        }
    }
    
    // Java对象到JSON转换
    public String objectToJson(Object obj) throws ConversionException {
        try {
            return objectMapper.writeValueAsString(obj);
        } catch (Exception e) {
            throw new ConversionException("对象转JSON失败", e);
        }
    }
    
    // XML到Java对象转换
    public <T> T xmlToObject(String xml, Class<T> clazz) throws ConversionException {
        try {
            Unmarshaller unmarshaller = jaxbContext.createUnmarshaller();
            return (T) unmarshaller.unmarshal(new StringReader(xml));
        } catch (Exception e) {
            throw new ConversionException("XML转换失败", e);
        }
    }
    
    // Java对象到XML转换
    public String objectToXml(Object obj) throws ConversionException {
        try {
            Marshaller marshaller = jaxbContext.createMarshaller();
            marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
            
            StringWriter writer = new StringWriter();
            marshaller.marshal(obj, writer);
            return writer.toString();
        } catch (Exception e) {
            throw new ConversionException("对象转XML失败", e);
        }
    }
    
    // 数据映射服务
    public <S, T> T mapData(S source, Class<T> targetClass) throws ConversionException {
        try {
            // 使用MapStruct进行高性能映射
            return dataMapper.map(source, targetClass);
        } catch (Exception e) {
            throw new ConversionException("数据映射失败", e);
        }
    }
    
    // 复杂数据转换
    public Map<String, Object> transformData(Map<String, Object> sourceData, 
        DataTransformationRule rule) throws ConversionException {
        
        Map<String, Object> targetData = new HashMap<>();
        
        // 应用转换规则
        for (FieldMapping mapping : rule.getFieldMappings()) {
            Object sourceValue = getValueByPath(sourceData, mapping.getSourcePath());
            
            // 应用转换函数
            Object targetValue = applyTransformation(sourceValue, mapping.getTransformation());
            
            // 设置目标值
            setValueByPath(targetData, mapping.getTargetPath(), targetValue);
        }
        
        return targetData;
    }
    
    private Object getValueByPath(Map<String, Object> data, String path) {
        String[] pathParts = path.split("\\.");
        Object currentValue = data;
        
        for (String part : pathParts) {
            if (currentValue instanceof Map) {
                currentValue = ((Map<String, Object>) currentValue).get(part);
            } else {
                return null;
            }
        }
        
        return currentValue;
    }
    
    private void setValueByPath(Map<String, Object> data, String path, Object value) {
        String[] pathParts = path.split("\\.");
        Map<String, Object> currentMap = data;
        
        for (int i = 0; i < pathParts.length - 1; i++) {
            String part = pathParts[i];
            if (!currentMap.containsKey(part)) {
                currentMap.put(part, new HashMap<String, Object>());
            }
            currentMap = (Map<String, Object>) currentMap.get(part);
        }
        
        currentMap.put(pathParts[pathParts.length - 1], value);
    }
    
    private Object applyTransformation(Object value, TransformationFunction function) {
        if (function == null || value == null) {
            return value;
        }
        
        switch (function.getType()) {
            case "DATE_FORMAT":
                return formatDate(value, function.getParameters().get("format"));
            case "STRING_UPPER":
                return value.toString().toUpperCase();
            case "STRING_LOWER":
                return value.toString().toLowerCase();
            case "NUMBER_ROUND":
                return roundNumber(value, function.getParameters().get("decimals"));
            case "CUSTOM":
                return applyCustomTransformation(value, function);
            default:
                return value;
        }
    }
}

错误处理与重试机制

完善的错误处理

// 集成错误处理服务
@Service
public class IntegrationErrorHandlingService {
    
    @Autowired
    private IntegrationErrorRepository errorRepository;
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private CircuitBreakerService circuitBreakerService;
    
    // 处理集成错误
    public void handleIntegrationError(IntegrationError error) {
        try {
            // 记录错误
            errorRepository.save(error);
            
            // 根据错误类型采取不同措施
            switch (error.getErrorType()) {
                case TRANSIENT:
                    handleTransientError(error);
                    break;
                case PERMANENT:
                    handlePermanentError(error);
                    break;
                case TIMEOUT:
                    handleTimeoutError(error);
                    break;
                default:
                    handleUnknownError(error);
            }
            
            // 发送通知
            notificationService.sendIntegrationErrorNotification(error);
        } catch (Exception e) {
            log.error("处理集成错误失败", e);
        }
    }
    
    // 处理临时性错误
    private void handleTransientError(IntegrationError error) {
        // 检查重试次数
        if (error.getRetryCount() < error.getMaxRetries()) {
            // 延迟重试
            scheduleRetry(error);
        } else {
            // 重试次数用尽，转为永久错误
            error.setErrorType(ErrorType.PERMANENT);
            handlePermanentError(error);
        }
    }
    
    // 处理永久性错误
    private void handlePermanentError(IntegrationError error) {
        // 记录到死信队列
        deadLetterQueueService.enqueue(error);
        
        // 触发人工干预流程
        triggerManualIntervention(error);
    }
    
    // 处理超时错误
    private void handleTimeoutError(IntegrationError error) {
        // 检查熔断器状态
        if (circuitBreakerService.isOpen(error.getServiceId())) {
            // 熔断器已打开，直接转为永久错误
            handlePermanentError(error);
        } else {
            // 尝试重试
            handleTransientError(error);
        }
    }
    
    // 调度重试
    private void scheduleRetry(IntegrationError error) {
        // 计算延迟时间（指数退避）
        long delay = calculateRetryDelay(error.getRetryCount());
        
        // 调度重试任务
        taskScheduler.schedule(() -> retryIntegration(error), 
            Instant.now().plusSeconds(delay));
    }
    
    // 计算重试延迟时间
    private long calculateRetryDelay(int retryCount) {
        // 指数退避算法：基础延迟 * 2^重试次数
        return 10 * (1L << retryCount); // 基础延迟10秒
    }
    
    // 重试集成
    private void retryIntegration(IntegrationError error) {
        try {
            // 增加重试次数
            error.setRetryCount(error.getRetryCount() + 1);
            
            // 重新执行集成操作
            boolean success = reexecuteIntegration(error);
            
            if (success) {
                // 重试成功，更新错误状态
                error.setStatus(ErrorStatus.RESOLVED);
                error.setResolvedTime(new Date());
                errorRepository.save(error);
                
                log.info("集成重试成功: {}", error.getRequestId());
            } else {
                // 重试失败，继续处理错误
                handleIntegrationError(error);
            }
        } catch (Exception e) {
            log.error("集成重试失败", e);
            error.setErrorMessage("重试失败: " + e.getMessage());
            handleIntegrationError(error);
        }
    }
}

重试机制实现

// 重试机制实现
@Component
public class RetryMechanism {
    
    // 带重试的集成调用
    public <T> T callWithRetry(IntegrationCall<T> call, RetryConfig config) 
        throws IntegrationException {
        
        Exception lastException = null;
        int attempt = 0;
        
        while (attempt <= config.getMaxAttempts()) {
            try {
                // 执行调用
                return call.execute();
            } catch (Exception e) {
                lastException = e;
                attempt++;
                
                // 检查是否需要重试
                if (attempt <= config.getMaxAttempts() && shouldRetry(e, config)) {
                    // 计算延迟时间
                    long delay = calculateDelay(attempt, config);
                    
                    // 等待后重试
                    try {
                        Thread.sleep(delay);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new IntegrationException("重试被中断", ie);
                    }
                } else {
                    // 不再重试，抛出异常
                    break;
                }
            }
        }
        
        // 所有重试都失败了
        throw new IntegrationException("集成调用失败，已重试" + config.getMaxAttempts() + "次", 
            lastException);
    }
    
    // 检查是否应该重试
    private boolean shouldRetry(Exception e, RetryConfig config) {
        // 检查异常类型
        if (config.getRetryableExceptions() != null) {
            for (Class<? extends Exception> retryableException : config.getRetryableExceptions()) {
                if (retryableException.isInstance(e)) {
                    return true;
                }
            }
            return false;
        }
        
        // 检查HTTP状态码（如果是HTTP调用）
        if (e instanceof HttpClientErrorException) {
            HttpClientErrorException httpError = (HttpClientErrorException) e;
            return config.getRetryableHttpStatuses().contains(httpError.getStatusCode().value());
        }
        
        // 默认情况下，临时性异常需要重试
        return e instanceof TransientIntegrationException;
    }
    
    // 计算延迟时间
    private long calculateDelay(int attempt, RetryConfig config) {
        long baseDelay = config.getBaseDelay();
        
        switch (config.getBackoffStrategy()) {
            case FIXED:
                return baseDelay;
            case LINEAR:
                return baseDelay * attempt;
            case EXPONENTIAL:
                return baseDelay * (1L << (attempt - 1)); // 2^(attempt-1)
            case EXPONENTIAL_WITH_JITTER:
                long exponentialDelay = baseDelay * (1L << (attempt - 1));
                // 添加抖动（±10%）
                double jitter = 0.1 * (Math.random() * 2 - 1);
                return (long) (exponentialDelay * (1 + jitter));
            default:
                return baseDelay;
        }
    }
    
    // 异步重试
    @Async
    public <T> CompletableFuture<T> callWithRetryAsync(IntegrationCall<T> call, 
        RetryConfig config) {
        
        return CompletableFuture.supplyAsync(() -> {
            try {
                return callWithRetry(call, config);
            } catch (IntegrationException e) {
                throw new CompletionException(e);
            }
        });
    }
}

安全认证与授权

认证机制

// 集成安全服务
@Service
public class IntegrationSecurityService {
    
    // OAuth2客户端凭证模式认证
    public String getOAuth2ClientCredentialsToken(OAuth2ClientConfig config) 
        throws SecurityException {
        
        try {
            // 构建认证请求
            MultiValueMap<String, String> params = new LinkedMultiValueMap<>();
            params.add("grant_type", "client_credentials");
            params.add("client_id", config.getClientId());
            params.add("client_secret", config.getClientSecret());
            params.add("scope", config.getScope());
            
            // 构建请求头
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
            headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
            
            HttpEntity<MultiValueMap<String, String>> request = 
                new HttpEntity<>(params, headers);
            
            // 发送认证请求
            RestTemplate restTemplate = new RestTemplate();
            ResponseEntity<OAuth2TokenResponse> response = restTemplate.postForEntity(
                config.getTokenUrl(), request, OAuth2TokenResponse.class);
            
            if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
                return response.getBody().getAccessToken();
            } else {
                throw new SecurityException("OAuth2认证失败: " + response.getStatusCode());
            }
        } catch (Exception e) {
            throw new SecurityException("获取OAuth2令牌失败", e);
        }
    }
    
    // JWT令牌验证
    public Claims validateJwtToken(String token, String publicKey) throws SecurityException {
        try {
            // 解析公钥
            RSAPublicKey rsaPublicKey = (RSAPublicKey) KeyFactory.getInstance("RSA")
                .generatePublic(new X509EncodedKeySpec(Base64.getDecoder().decode(publicKey)));
            
            // 验证JWT令牌
            JwtParser parser = Jwts.parserBuilder()
                .setSigningKey(rsaPublicKey)
                .build();
            
            return parser.parseClaimsJws(token).getBody();
        } catch (Exception e) {
            throw new SecurityException("JWT令牌验证失败", e);
        }
    }
    
    // API密钥认证
    public boolean validateApiKey(String apiKey, String serviceId) {
        try {
            // 从数据库或配置中心获取API密钥信息
            ApiKeyInfo apiKeyInfo = apiKeyRepository.findByApiKeyAndServiceId(apiKey, serviceId);
            
            if (apiKeyInfo == null) {
                return false;
            }
            
            // 检查是否过期
            if (apiKeyInfo.getExpireTime() != null && 
                apiKeyInfo.getExpireTime().before(new Date())) {
                return false;
            }
            
            // 检查是否被禁用
            if (!apiKeyInfo.isEnabled()) {
                return false;
            }
            
            // 记录使用情况
            recordApiKeyUsage(apiKeyInfo);
            
            return true;
        } catch (Exception e) {
            log.error("API密钥验证失败", e);
            return false;
        }
    }
    
    // 记录API密钥使用情况
    private void recordApiKeyUsage(ApiKeyInfo apiKeyInfo) {
        try {
            ApiKeyUsage usage = new ApiKeyUsage();
            usage.setApiKey(apiKeyInfo.getApiKey());
            usage.setServiceId(apiKeyInfo.getServiceId());
            usage.setUseTime(new Date());
            usage.setIpAddress(getClientIpAddress());
            
            apiKeyUsageRepository.save(usage);
        } catch (Exception e) {
            log.warn("记录API密钥使用情况失败", e);
        }
    }
}

案例分析

案例一：制造业的ERP系统集成

某制造企业通过BPM平台实现了与ERP系统的深度集成：

集成需求

订单管理：自动同步销售订单到ERP系统
库存管理：实时获取库存信息用于生产计划
财务管理：自动生成财务凭证和报表
采购管理：根据生产计划自动创建采购订单

技术实现

// ERP系统集成服务
@Service
public class ERPIntegrationService {
    
    // 同步销售订单到ERP
    public boolean syncSalesOrder(SalesOrder order) {
        try {
            // 转换数据格式
            ERPSalesOrder erpOrder = dataMapper.mapToERPOrder(order);
            
            // 调用ERP API
            ERPApiResponse response = erpApiClient.createSalesOrder(erpOrder);
            
            if (response.isSuccess()) {
                // 更新订单状态
                order.setErpOrderId(response.getOrderId());
                order.setStatus(OrderStatus.SYNCED);
                salesOrderRepository.save(order);
                
                log.info("销售订单同步成功: {}", order.getOrderNo());
                return true;
            } else {
                log.error("销售订单同步失败: {}, 错误: {}", 
                    order.getOrderNo(), response.getErrorMessage());
                return false;
            }
        } catch (Exception e) {
            log.error("销售订单同步异常: {}", order.getOrderNo(), e);
            return false;
        }
    }
    
    // 获取库存信息
    public List<InventoryInfo> getInventoryInfo(List<String> productCodes) {
        try {
            // 构建请求参数
            InventoryQueryRequest request = new InventoryQueryRequest();
            request.setProductCodes(productCodes);
            request.setTimestamp(new Date());
            
            // 调用ERP库存查询API
            ERPApiResponse response = erpApiClient.queryInventory(request);
            
            if (response.isSuccess()) {
                return dataMapper.mapToInventoryInfo(response.getData());
            } else {
                throw new IntegrationException("库存查询失败: " + response.getErrorMessage());
            }
        } catch (Exception e) {
            log.error("库存查询异常", e);
            throw new IntegrationException("库存查询失败", e);
        }
    }
    
    // 处理ERP回调
    @EventListener
    public void handleERPCallback(ERPCallbackEvent event) {
        try {
            switch (event.getEventType()) {
                case "ORDER_STATUS_CHANGED":
                    updateOrderStatus(event.getData());
                    break;
                case "INVENTORY_UPDATED":
                    updateInventoryCache(event.getData());
                    break;
                case "PAYMENT_RECEIVED":
                    processPayment(event.getData());
                    break;
                default:
                    log.warn("未知的ERP回调事件类型: {}", event.getEventType());
            }
        } catch (Exception e) {
            log.error("处理ERP回调事件失败", e);
            // 可以发送到死信队列进行后续处理
        }
    }
}

实施效果

订单处理时间减少60%
库存准确率提升至99.8%
财务对账效率提升75%
人工干预减少80%

案例二：金融行业的多系统集成

某银行通过BPM平台实现了核心银行系统、征信系统、风险管理系统等多个系统的集成：

集成架构

// 银行系统集成网关
@Component
public class BankIntegrationGateway {
    
    // 信贷申请处理流程
    public CreditApplicationResult processCreditApplication(CreditApplication application) {
        CreditApplicationResult result = new CreditApplicationResult();
        result.setApplicationId(application.getId());
        result.setStartTime(new Date());
        
        try {
            // 1. 客户信息验证
            CustomerInfo customerInfo = validateCustomerInfo(application.getCustomerId());
            result.setCustomerValidated(true);
            
            // 2. 征信查询
            CreditReport creditReport = queryCreditReport(application.getCustomerId());
            result.setCreditReport(creditReport);
            
            // 3. 风险评估
            RiskAssessmentResult riskResult = assessRisk(application, customerInfo, creditReport);
            result.setRiskAssessment(riskResult);
            
            // 4. 额度计算
            CreditLimit creditLimit = calculateCreditLimit(application, riskResult);
            result.setCreditLimit(creditLimit);
            
            // 5. 核心系统处理
            CoreBankingResult coreResult = processInCoreBanking(application, creditLimit);
            result.setCoreBankingResult(coreResult);
            
            // 6. 结果通知
            notifyApplicationResult(application, result);
            
            result.setStatus(ApplicationStatus.APPROVED);
        } catch (Exception e) {
            log.error("信贷申请处理失败: {}", application.getId(), e);
            result.setStatus(ApplicationStatus.REJECTED);
            result.setErrorMessage(e.getMessage());
            
            // 记录错误并触发人工处理
            triggerManualReview(application, e);
        } finally {
            result.setEndTime(new Date());
            result.setDuration(result.getEndTime().getTime() - result.getStartTime().getTime());
        }
        
        return result;
    }
    
    // 客户信息验证
    private CustomerInfo validateCustomerInfo(String customerId) throws IntegrationException {
        try {
            // 调用核心银行系统验证客户信息
            CustomerInfo customerInfo = coreBankingService.getCustomerInfo(customerId);
            
            if (customerInfo == null) {
                throw new IntegrationException("客户信息不存在: " + customerId);
            }
            
            // 验证客户状态
            if (!"ACTIVE".equals(customerInfo.getStatus())) {
                throw new IntegrationException("客户状态异常: " + customerInfo.getStatus());
            }
            
            return customerInfo;
        } catch (Exception e) {
            throw new IntegrationException("客户信息验证失败", e);
        }
    }
    
    // 征信查询
    private CreditReport queryCreditReport(String customerId) throws IntegrationException {
        try {
            // 调用征信系统查询信用报告
            CreditQueryRequest request = new CreditQueryRequest();
            request.setCustomerId(customerId);
            request.setQueryType("FULL");
            request.setTimestamp(new Date());
            
            CreditQueryResponse response = creditBureauService.queryCredit(request);
            
            if (!response.isSuccess()) {
                throw new IntegrationException("征信查询失败: " + response.getErrorMessage());
            }
            
            return response.getCreditReport();
        } catch (Exception e) {
            throw new IntegrationException("征信查询异常", e);
        }
    }
    
    // 风险评估
    private RiskAssessmentResult assessRisk(CreditApplication application, 
        CustomerInfo customerInfo, CreditReport creditReport) throws IntegrationException {
        
        try {
            // 构建风险评估输入
            RiskAssessmentInput input = new RiskAssessmentInput();
            input.setApplication(application);
            input.setCustomerInfo(customerInfo);
            input.setCreditReport(creditReport);
            
            // 调用风险管理系统进行评估
            return riskManagementService.assessCreditRisk(input);
        } catch (Exception e) {
            throw new IntegrationException("风险评估失败", e);
        }
    }
}

业务效果

信贷审批时间从3天缩短到30分钟
审批准确率提升至99.5%
风险控制能力显著增强
客户满意度提升30%

最佳实践总结

集成设计原则

松耦合设计
- 通过接口抽象屏蔽系统差异
- 实现系统间的松耦合
- 支持系统的独立演进
容错性设计
- 实现完善的错误处理机制
- 支持重试和熔断机制
- 提供降级处理方案
可观测性
- 建立完整的监控指标体系
- 实现端到端的追踪能力
- 提供详细的日志记录

性能优化建议

连接池管理
- 合理配置连接池参数
- 实现连接的复用和管理
- 监控连接池使用情况
缓存策略
- 对频繁访问的数据实施缓存
- 合理设置缓存过期时间
- 实现缓存的更新机制
异步处理
- 对非实时性要求的操作采用异步处理
- 使用消息队列实现系统解耦
- 实现批量处理提升效率

结语

服务调用与集成是BPM平台实现端到端业务流程自动化的核心能力。通过合理的设计和实现，我们可以构建出高效、可靠、安全的系统集成体系，打通企业内外部系统的数据壁垒，实现业务流程的无缝衔接和自动化执行。

在实际项目实施中，我们需要根据具体的业务需求和技术环境，选择合适的集成模式和技术方案。同时，要重视集成的安全性、可靠性和可维护性，建立完善的监控和运维体系，确保集成系统能够稳定、高效地支撑业务流程的执行。

随着云原生、微服务、API经济等技术趋势的发展，系统集成正朝着更加标准化、平台化、智能化的方向演进。未来的BPM平台将需要具备更强的集成能力，支持更多样化的集成模式，为企业数字化转型提供更加有力的技术支撑。