## 一、模式概述

管道模式和责任链模式是常用的行为型设计模式：

**管道模式（Pipeline Pattern）：**
- 数据像水一样在管道中流动
- 每个管道处理一个特定的任务
- 管道可以串联和并联

**责任链模式（Chain of Responsibility）：**
- 请求在链中传递
- 每个处理器决定处理或传递
- 解耦发送者和接收者

## 二、管道模式

### 1. 核心概念

```
┌─────────────────────────────────────────────────────────────────┐
│                        管道模式                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────┐    ┌─────┐    ┌─────┐    ┌─────┐    ┌─────┐        │
│  │输入 │───▶│管道1 │───▶│管道2 │───▶│管道3 │───▶│输出 │        │
│  └─────┘    └─────┘    └─────┘    └─────┘    └─────┘        │
│                  │            │            │                     │
│               处理A         处理B         处理C                   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
```

### 2. 管道接口定义

```java
// 管道上下文，携带数据在管道中流动
public class PipelineContext {
    private Map<String, Object> attributes = new ConcurrentHashMap<>();
    private Object data;
    private List<String> executedHandlers = new ArrayList<>();
    
    public void setAttribute(String key, Object value) {
        attributes.put(key, value);
    }
    
    public Object getAttribute(String key) {
        return attributes.get(key);
    }
    
    public Object getData() {
        return data;
    }
    
    public void setData(Object data) {
        this.data = data;
    }
    
    public void addExecutedHandler(String handler) {
        executedHandlers.add(handler);
    }
    
    public List<String> getExecutedHandlers() {
        return executedHandlers;
    }
}

// 管道处理器接口
public interface PipelineHandler<T> {
    
    // 处理器名称
    String getName();
    
    // 处理数据
    void handle(T data, PipelineContext context);
    
    // 是否异步执行
    default boolean isAsync() {
        return false;
    }
}

// 管道接口
public interface Pipeline<T> {
    
    Pipeline<T> addHandler(PipelineHandler<T> handler);
    
    Pipeline<T> addHandler(int index, PipelineHandler<T> handler);
    
    void removeHandler(String name);
    
    void execute(T data);
    
    void executeAsync(T data);
}
```

### 3. 管道实现

```java
// 管道实现
public class DefaultPipeline<T> implements Pipeline<T> {
    
    private final List<PipelineHandler<T>> handlers = new ArrayList<>();
    private final PipelineContext context = new PipelineContext();
    private PipelineExecutor executor;
    
    public DefaultPipeline(PipelineExecutor executor) {
        this.executor = executor;
    }
    
    @Override
    public Pipeline<T> addHandler(PipelineHandler<T> handler) {
        handlers.add(handler);
        return this;
    }
    
    @Override
    public Pipeline<T> addHandler(int index, PipelineHandler<T> handler) {
        handlers.add(index, handler);
        return this;
    }
    
    @Override
    public void removeHandler(String name) {
        handlers.removeIf(h -> h.getName().equals(name));
    }
    
    @Override
    public void execute(T data) {
        context.setData(data);
        executeInternal(0);
    }
    
    @Override
    public void executeAsync(T data) {
        context.setData(data);
        executor.executeAsync(() -> executeInternal(0));
    }
    
    private void executeInternal(int index) {
        if (index >= handlers.size()) {
            return;
        }
        
        PipelineHandler<T> handler = handlers.get(index);
        
        try {
            handler.handle(data, context);
            context.addExecutedHandler(handler.getName());
            
            if (handler.isAsync()) {
                // 异步处理器需要主动调用下一个
                executeInternal(index + 1);
            } else {
                // 同步处理器继续执行下一个
                executeInternal(index + 1);
            }
        } catch (Exception e) {
            handleError(handler, e, index);
        }
    }
    
    private void handleError(PipelineHandler<T> handler, Exception e, int index) {
        // 错误处理逻辑
        if (handler instanceof ErrorHandler) {
            ((ErrorHandler) handler).handleError(e, context);
        }
    }
}

// 管道工厂
@Component
public class PipelineFactory {
    
    private final Map<String, Pipeline<?>> pipelines = new ConcurrentHashMap<>();
    
    public <T> Pipeline<T> getPipeline(String name) {
        return (Pipeline<T>) pipelines.get(name);
    }
    
    public <T> void registerPipeline(String name, Pipeline<T> pipeline) {
        pipelines.put(name, pipeline);
    }
}
```

## 三、责任链模式

### 1. 核心概念

```
┌─────────────────────────────────────────────────────────────────┐
│                      责任链模式                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────┐    ┌─────┐    ┌─────┐    ┌─────┐                      │
│  │处理器1│──▶│处理器2│──▶│处理器3│──▶│处理器4│───▶ 结束       │
│  └─────┘    └─────┘    └─────┘    └─────┘                      │
│       │          │          │          │                          │
│    处理      处理并传递    处理并传递    不处理                      │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
```

### 2. 责任链实现

```java
// 抽象处理器
public abstract class Handler<T> {
    
    private Handler<T> next;
    
    public Handler<T> setNext(Handler<T> next) {
        this.next = next;
        return next;
    }
    
    public final void handle(T request) {
        if (canHandle(request)) {
            doHandle(request);
        } else if (next != null) {
            next.handle(request);
        } else {
            onUnhandled(request);
        }
    }
    
    protected abstract boolean canHandle(T request);
    
    protected abstract void doHandle(T request);
    
    protected void onUnhandled(T request) {
        // 默认不处理
    }
}

// 认证处理器
@Component
public class AuthHandler extends Handler<Request> {
    
    @Override
    protected boolean canHandle(Request request) {
        return request.getAuthToken() == null;
    }
    
    @Override
    protected void doHandle(Request request) {
        throw new AuthException("未认证");
    }
}

// 权限处理器
@Component
public class PermissionHandler extends Handler<Request> {
    
    @Override
    protected boolean canHandle(Request request) {
        return !hasPermission(request);
    }
    
    @Override
    protected void doHandle(Request request) {
        throw new ForbiddenException("无权限");
    }
    
    private boolean hasPermission(Request request) {
        // 检查权限逻辑
        return true;
    }
}

// 参数校验处理器
@Component
public class ValidationHandler extends Handler<Request> {
    
    @Override
    protected boolean canHandle(Request request) {
        return !validate(request);
    }
    
    @Override
    protected void doHandle(Request request) {
        throw new ValidationException("参数校验失败");
    }
    
    private boolean validate(Request request) {
        return request != null && request.getData() != null;
    }
}
```

## 四、实战案例

### 1. 请求处理管道

```java
// Web请求处理管道
@Component
public class WebRequestPipeline {
    
    @Autowired
    private List<RequestHandler> handlers;
    
    @Autowired
    private PipelineFactory pipelineFactory;
    
    @PostConstruct
    public void init() {
        Pipeline<HttpRequest> pipeline = pipelineFactory.getPipeline("web-request");
        
        handlers.stream()
            .sorted(Comparator.comparingInt(RequestHandler::getOrder))
            .forEach(pipeline::addHandler);
    }
}

// 请求处理器
@Component
public class AuthRequestHandler implements RequestHandler {
    
    @Override
    public int getOrder() {
        return 1;
    }
    
    @Override
    public void handle(HttpRequest data, PipelineContext context) {
        String token = data.getHeader("Authorization");
        
        if (token == null) {
            throw new UnauthorizedException("未登录");
        }
        
        // 验证Token
        UserPrincipal principal = authService.verifyToken(token);
        context.setAttribute("user", principal);
    }
}

@Component
public class LogRequestHandler implements RequestHandler {
    
    @Override
    public int getOrder() {
        return 100;
    }
    
    @Override
    public void handle(HttpRequest data, PipelineContext context) {
        long start = System.currentTimeMillis();
        context.setAttribute("startTime", start);
        
        // 记录请求日志
        log.info("请求开始: {} {}", data.getMethod(), data.getPath());
    }
}
```

### 2. 数据处理管道

```java
// 数据处理管道
@Component
public class DataProcessPipeline {
    
    public List<ProcessedData> process(List<RawData> rawDataList) {
        Pipeline<RawData> pipeline = createPipeline();
        
        List<ProcessedData> results = new ArrayList<>();
        
        for (RawData raw : rawDataList) {
            PipelineContext context = new PipelineContext();
            pipeline.execute(raw);
            
            ProcessedData result = (ProcessedData) context.getAttribute("result");
            results.add(result);
        }
        
        return results;
    }
    
    private Pipeline<RawData> createPipeline() {
        return new DefaultPipeline<>(new SyncPipelineExecutor())
            .addHandler(new ParseHandler())
            .addHandler(new CleanHandler())
            .addHandler(new TransformHandler())
            .addHandler(new EnrichHandler())
            .addHandler(new ValidateHandler())
            .addHandler(new AggregateHandler());
    }
}

// 解析处理器
public class ParseHandler implements PipelineHandler<RawData> {
    
    @Override
    public String getName() {
        return "parse";
    }
    
    @Override
    public void handle(RawData data, PipelineContext context) {
        try {
            Object parsed = JSON.parse(data.getContent());
            context.setAttribute("parsed", parsed);
        } catch (Exception e) {
            throw new ProcessException("解析失败: " + e.getMessage());
        }
    }
}

// 清洗处理器
public class CleanHandler implements PipelineHandler<RawData> {
    
    @Override
    public String getName() {
        return "clean";
    }
    
    @Override
    public void handle(RawData data, PipelineContext context) {
        Object parsed = context.getAttribute("parsed");
        Object cleaned = cleanData(parsed);
        context.setAttribute("cleaned", cleaned);
    }
    
    private Object cleanData(Object data) {
        // 清洗数据：去除空白、规范化格式等
        return data;
    }
}
```

### 3. 消息处理责任链

```java
// 消息处理责任链
@Component
public class MessageHandlerChain {
    
    private Handler<Message> chain;
    
    @Autowired
    private List<MessageHandler> handlers;
    
    @PostConstruct
    public void init() {
        // 按优先级排序
        handlers.sort(Comparator.comparingInt(MessageHandler::getPriority));
        
        // 构建责任链
        chain = new AuthMessageHandler();
        Handler<Message> current = chain;
        
        for (MessageHandler handler : handlers) {
            current = current.setNext(handler);
        }
    }
    
    public void handle(Message message) {
        chain.handle(message);
    }
}

// 消息处理器
@Component
public class AuthMessageHandler extends Handler<Message> {
    
    @Override
    protected boolean canHandle(Message message) {
        return message.getAuthToken() == null;
    }
    
    @Override
    protected void doHandle(Message message) {
        throw new AuthException("消息认证失败");
    }
}

@Component
public class RateLimitMessageHandler extends Handler<Message> {
    
    @Override
    protected boolean canHandle(Message message) {
        return !rateLimiter.tryAcquire(message.getSenderId());
    }
    
    @Override
    protected void doHandle(Message message) {
        throw new RateLimitException("发送频率超限");
    }
}
```

## 五、过滤器模式

### 1. Servlet过滤器

```java
// 过滤器链
@Component
@Order(1)
public class AuthFilter implements Filter {
    
    @Override
    public void doFilter(ServletRequest request, ServletResponse response, 
                         FilterChain chain) throws IOException, ServletException {
        HttpServletRequest httpRequest = (HttpServletRequest) request;
        
        String token = httpRequest.getHeader("Authorization");
        
        if (token == null || !authService.validate(token)) {
            HttpServletResponse httpResponse = (HttpServletResponse) response;
            httpResponse.setStatus(401);
            httpResponse.getWriter().write("Unauthorized");
            return;
        }
        
        chain.doFilter(request, response);
    }
}

@Component
@Order(2)
public class LoggingFilter implements Filter {
    
    @Override
    public void doFilter(ServletRequest request, ServletResponse response, 
                         FilterChain chain) throws IOException, ServletException {
        long start = System.currentTimeMillis();
        
        chain.doFilter(request, response);
        
        long duration = System.currentTimeMillis() - start;
        HttpServletRequest httpRequest = (HttpServletRequest) request;
        
        log.info("{} {} - {}ms", 
            httpRequest.getMethod(), 
            httpRequest.getRequestURI(), 
            duration);
    }
}
```

### 2. Spring拦截器

```java
@Configuration
public class WebMvcConfig implements WebMvcConfigurer {
    
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new AuthInterceptor())
            .addPathPatterns("/api/**")
            .excludePathPatterns("/api/public/**");
        
        registry.addInterceptor(new LoggingInterceptor())
            .addPathPatterns("/**");
    }
}

@Component
public class AuthInterceptor extends HandlerInterceptorAdapter {
    
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, 
                            Object handler) throws Exception {
        String token = request.getHeader("Authorization");
        
        if (token == null || !authService.validate(token)) {
            response.setStatus(401);
            response.getWriter().write("Unauthorized");
            return false;
        }
        
        return true;
    }
}
```

## 六、异步管道

### 1. 异步执行

```java
// 异步管道执行器
@Component
public class AsyncPipelineExecutor implements PipelineExecutor {
    
    private final ExecutorService executor = Executors.newCachedThreadPool();
    
    @Override
    public void execute(Runnable task) {
        executor.submit(task);
    }
    
    @Override
    public <T> CompletableFuture<T> executeAsync(Supplier<T> supplier) {
        return CompletableFuture.supplyAsync(supplier, executor);
    }
}

// 异步处理器
@Component
public class AsyncHandler implements PipelineHandler<Request> {
    
    @Override
    public String getName() {
        return "async";
    }
    
    @Override
    public boolean isAsync() {
        return true;
    }
    
    @Override
    public void handle(Request data, PipelineContext context) {
        CompletableFuture.runAsync(() -> {
            // 异步处理
            processAsync(data);
            
            // 通知管道继续
            synchronized (context) {
                context.notify();
            }
        });
        
        // 等待异步处理完成
        synchronized (context) {
            try {
                context.wait();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}
```

### 2. 并行处理

```java
// 分支管道
@Component
public class BranchPipeline<T> {
    
    private final List<Pipeline<T>> branches = new ArrayList<>();
    
    public BranchPipeline<T> addBranch(Pipeline<T> pipeline) {
        branches.add(pipeline);
        return this;
    }
    
    public void execute(T data) {
        List<CompletableFuture<Void>> futures = branches.stream()
            .map(pipeline -> CompletableFuture.runAsync(() -> pipeline.execute(data)))
            .collect(Collectors.toList());
        
        // 等待所有分支完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }
}

// 使用示例
public class MultiProcessPipeline {
    
    public ProcessedResult process(RawData raw) {
        BranchPipeline<RawData> branch = new BranchPipeline<RawData>()
            .addBranch(createValidationPipeline())
            .addBranch(createEnrichmentPipeline())
            .addBranch(createAggregationPipeline());
        
        branch.execute(raw);
        
        return mergeResults();
    }
}
```

## 七、模式对比与选择

### 1. 管道 vs 责任链

| 维度 | 管道模式 | 责任链模式 |
|------|---------|-----------|
| 数据流 | 顺序流动 | 选择性传递 |
| 处理 | 所有处理器都执行 | 找到处理者即停止 |
| 用途 | 数据处理流水线 | 请求拦截过滤 |
| 灵活性 | 高，可动态编排 | 中，按序执行 |

### 2. 适用场景

```
管道模式适用：
- 数据 ETL 处理
- 请求/响应处理
- 图片/文件处理
- 多步骤业务流程

责任链模式适用：
- 认证授权
- 日志记录
- 错误处理
- 插件系统
```

## 八、总结

管道模式和责任链模式是处理流程的利器：

- **管道模式**：数据顺序处理，适用于多步骤流程
- **责任链模式**：请求选择处理，适用于拦截过滤
- **组合使用**：可以构建复杂的处理流程
- **异步支持**：提升处理性能

**最佳实践：**
1. 明确定义处理器接口
2. 做好异常处理
3. 支持异步执行
4. 合理设置处理器顺序

---

*个人观点，仅供参考*

【架构实战】管道模式与责任链模式实战

研发部经理

开发工具

管道模式与责任链模式是两种常见的行为型设计模式。管道模式适用于顺序处理数据的场景，如ETL流程或请求处理流水线，确保每个步骤依次执行。责任链模式则适合拦截过滤型任务，如认证授权、日志记录等，请求在链中传递并由符合条件的处理器处理。两者均可提升代码灵活性与可维护性，支持异步执行优化性能。

数据流

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【架构实战】管道模式与责任链模式实战-腾讯云开发者社区-腾讯云

【架构实战】管道模式与责任链模式实战

【架构实战】管道模式与责任链模式实战

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐