Java数据管道案例

wen java案例 2026-07-30 247

本文目录导读：

Java数据管道案例

基础数据处理管道（Stream API）
使用Pipeline模式（设计模式）
实时数据处理管道（有界队列）
大数据处理管道（批量处理）
使用建议

我来为您介绍几个Java数据管道的典型案例,涵盖不同的使用场景和实现方式。

基础数据处理管道（Stream API）

import java.util.*;
import java.util.stream.*;
public class DataPipelineExample {
    public static void main(String[] args) {
        // 模拟数据源
        List<Transaction> transactions = Arrays.asList(
            new Transaction("T001", "Alice", 1500.0, "2024-01-15"),
            new Transaction("T002", "Bob", 3200.0, "2024-01-16"),
            new Transaction("T003", "Alice", 800.0, "2024-01-17"),
            new Transaction("T004", "Charlie", 5000.0, "2024-01-18"),
            new Transaction("T005", "Bob", 1200.0, "2024-01-19")
        );
        // 数据管道：过滤 -> 转换 -> 聚合
        Map<String, Double> result = transactions.stream()
            .filter(t -> t.getAmount() > 1000)  // 过滤大额交易
            .peek(t -> System.out.println("Filtered: " + t))  // 调试
            .collect(Collectors.groupingBy(
                Transaction::getPerson,
                Collectors.summingDouble(Transaction::getAmount)
            ));
        System.out.println("每人总交易额（>1000）：" + result);
    }
}
class Transaction {
    private String id;
    private String person;
    private double amount;
    private String date;
    // 构造函数、getter/setter省略
    public Transaction(String id, String person, double amount, String date) {
        this.id = id;
        this.person = person;
        this.amount = amount;
        this.date = date;
    }
    public String getPerson() { return person; }
    public double getAmount() { return amount; }
    @Override
    public String toString() {
        return String.format("Transaction{id='%s', amount=%.2f}", id, amount);
    }
}

使用Pipeline模式（设计模式）

import java.util.function.Function;
// 管道处理器接口
interface PipelineHandler<T, R> {
    R process(T input);
}
// 管道构建器
class Pipeline<T, R> {
    private final Function<T, R> currentHandler;
    private Pipeline(Function<T, R> handler) {
        this.currentHandler = handler;
    }
    public static <T> Pipeline<T, T> create() {
        return new Pipeline<>(Function.identity());
    }
    public <V> Pipeline<T, V> addHandler(PipelineHandler<R, V> handler) {
        return new Pipeline<>(currentHandler.andThen(handler::process));
    }
    public R execute(T input) {
        return currentHandler.apply(input);
    }
}
// 具体处理器
class ValidateHandler implements PipelineHandler<String, String> {
    @Override
    public String process(String input) {
        if (input == null || input.trim().isEmpty()) {
            throw new IllegalArgumentException("Input cannot be empty");
        }
        System.out.println("Validation passed: " + input);
        return input.trim();
    }
}
class TransformHandler implements PipelineHandler<String, String> {
    @Override
    public String process(String input) {
        String result = input.toUpperCase();
        System.out.println("Transformed: " + result);
        return result;
    }
}
class FormatHandler implements PipelineHandler<String, String> {
    @Override
    public String process(String input) {
        String result = "[" + input + "]";
        System.out.println("Formatted: " + result);
        return result;
    }
}
// 使用示例
public class PipelinePatternExample {
    public static void main(String[] args) {
        Pipeline<String, String> textPipeline = Pipeline.<String>create()
            .addHandler(new ValidateHandler())
            .addHandler(new TransformHandler())
            .addHandler(new FormatHandler());
        String result = textPipeline.execute("  hello world  ");
        System.out.println("Final result: " + result);
    }
}

实时数据处理管道（有界队列）

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
class DataProcessor {
    private final BlockingQueue<String> inputQueue;
    private final BlockingQueue<String> outputQueue;
    private final AtomicInteger processedCount = new AtomicInteger(0);
    public DataProcessor(int capacity) {
        this.inputQueue = new LinkedBlockingQueue<>(capacity);
        this.outputQueue = new LinkedBlockingQueue<>(capacity);
    }
    // 数据源
    public void produceData(String data) throws InterruptedException {
        inputQueue.put(data);
        System.out.println("Produced: " + data);
    }
    // 处理阶段
    public void processData() {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        // 阶段1：清理数据
        executor.submit(() -> {
            try {
                while (true) {
                    String data = inputQueue.take();
                    String cleaned = data.trim().toLowerCase();
                    System.out.println("Stage1 - Cleaned: " + cleaned);
                    // 传递给下一阶段
                    String processed = cleaned;
                    executor.submit(() -> {
                        try {
                            // 阶段2：转换数据
                            Thread.sleep(100); // 模拟处理时间
                            String transformed = "processed_" + processed;
                            System.out.println("Stage2 - Transformed: " + transformed);
                            // 阶段3：输出结果
                            outputQueue.put(transformed);
                            int count = processedCount.incrementAndGet();
                            System.out.println("Complete - Total processed: " + count);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    });
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
    public String consumeData() throws InterruptedException {
        return outputQueue.take();
    }
    public int getProcessedCount() {
        return processedCount.get();
    }
}
// 使用示例
public class RealTimePipelineExample {
    public static void main(String[] args) throws InterruptedException {
        DataProcessor processor = new DataProcessor(100);
        // 启动处理管道
        processor.processData();
        // 生产数据
        String[] testData = {"  Data1  ", "  Data2  ", "  Data3  "};
        for (String data : testData) {
            processor.produceData(data);
        }
        // 消费结果
        Thread.sleep(1000); // 等待处理完成
        System.out.println("Processed count: " + processor.getProcessedCount());
    }
}

大数据处理管道（批量处理）

import java.util.*;
import java.util.stream.Collectors;
// 数据记录
class DataRecord {
    private String category;
    private double value;
    private long timestamp;
    public DataRecord(String category, double value, long timestamp) {
        this.category = category;
        this.value = value;
        this.timestamp = timestamp;
    }
    // getter方法
    public String getCategory() { return category; }
    public double getValue() { return value; }
    public long getTimestamp() { return timestamp; }
}
// 管道阶段接口
interface PipelineStage<T> {
    List<T> process(List<T> input);
}
// 具体处理阶段
class FilterStage implements PipelineStage<DataRecord> {
    private double minValue;
    public FilterStage(double minValue) {
        this.minValue = minValue;
    }
    @Override
    public List<DataRecord> process(List<DataRecord> input) {
        return input.stream()
            .filter(r -> r.getValue() > minValue)
            .collect(Collectors.toList());
    }
}
class TransformStage implements PipelineStage<DataRecord> {
    private double multiplier;
    public TransformStage(double multiplier) {
        this.multiplier = multiplier;
    }
    @Override
    public List<DataRecord> process(List<DataRecord> input) {
        return input.stream()
            .peek(r -> new DataRecord(r.getCategory(), r.getValue() * multiplier, r.getTimestamp()))
            .collect(Collectors.toList());
    }
}
class AggregateStage implements PipelineStage<DataRecord> {
    @Override
    public List<DataRecord> process(List<DataRecord> input) {
        // 按类别聚合
        Map<String, Double> aggregated = input.stream()
            .collect(Collectors.groupingBy(
                DataRecord::getCategory,
                Collectors.summingDouble(DataRecord::getValue)
            ));
        return aggregated.entrySet().stream()
            .map(e -> new DataRecord(e.getKey(), e.getValue(), System.currentTimeMillis()))
            .collect(Collectors.toList());
    }
}
// 管道管理器
class DataPipelineManager {
    private List<PipelineStage<DataRecord>> stages = new ArrayList<>();
    public DataPipelineManager addStage(PipelineStage<DataRecord> stage) {
        stages.add(stage);
        return this;
    }
    public List<DataRecord> execute(List<DataRecord> data) {
        List<DataRecord> current = new ArrayList<>(data);
        for (PipelineStage<DataRecord> stage : stages) {
            System.out.println("Processing stage: " + stage.getClass().getSimpleName());
            current = stage.process(current);
            System.out.println("Stage output size: " + current.size());
        }
        return current;
    }
}
// 使用示例
public class BatchProcessingPipelineExample {
    public static void main(String[] args) {
        // 准备数据
        List<DataRecord> data = Arrays.asList(
            new DataRecord("A", 100.0, System.currentTimeMillis()),
            new DataRecord("B", 200.0, System.currentTimeMillis()),
            new DataRecord("A", 150.0, System.currentTimeMillis()),
            new DataRecord("C", 50.0, System.currentTimeMillis()),
            new DataRecord("B", 300.0, System.currentTimeMillis())
        );
        // 构建管道
        DataPipelineManager pipeline = new DataPipelineManager()
            .addStage(new FilterStage(80.0))
            .addStage(new TransformStage(1.1))
            .addStage(new AggregateStage());
        // 执行管道
        List<DataRecord> result = pipeline.execute(data);
        System.out.println("Final Results:");
        result.forEach(r -> 
            System.out.printf("Category: %s, Total Value: %.2f%n", 
                r.getCategory(), r.getValue())
        );
    }
}

使用建议

选择合适的管道实现：
- 小数据量：使用Stream API
- 复杂业务逻辑：使用Pipeline设计模式
- 实时处理：使用BlockingQueue
- 大数据量：使用批量处理框架
性能考虑：
- 使用并行流处理大数据
- 合理设置缓冲区大小
- 考虑使用CompletableFuture实现异步
错误处理：
- 实现重试机制
- 添加日志记录
- 使用熔断器模式

这些案例覆盖了不同场景下的数据管道实现,您可以根据实际需求选择合适的方案。

本文地址： https://www.bmjup.com/post/34143.html