本文目录导读：
我来为您介绍几个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实现异步
-
错误处理：
- 实现重试机制
- 添加日志记录
- 使用熔断器模式
这些案例覆盖了不同场景下的数据管道实现,您可以根据实际需求选择合适的方案。
