如何使用ChannelDuplexHandler替代SimpleChannelInboundHandler并分离读写逻辑

Got it, let's break this down step by step. You want to split the read and write logic from your original SimpleChannelInboundHandler into separate concerns using Netty's handler hierarchy—either splitting into dedicated inbound/outbound handlers or using ChannelDuplexHandler to encapsulate both. Let's walk through practical implementations for both approaches.

Option 1: Split into Dedicated Inbound & Outbound Handlers (Single Responsibility) #

This is the cleaner approach, as it separates request processing (inbound) from response writing logic (outbound).

Step 1: Inbound Handler for Request Processing #

This handler will only handle incoming RpcRequest messages, process them to generate a RpcResponse, and pass the response to the outbound pipeline.

import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RpcRequestInboundHandler extends SimpleChannelInboundHandler<RpcRequest> {
    private static final Logger logger = LoggerFactory.getLogger(RpcRequestInboundHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        logger.debug("Received request: {}", request);
        RpcResponse response = new RpcResponse(request.getRequestId());
        
        try {
            Object result = Server.handleRequest(request);
            response.setResult(result);
        } catch (Throwable throwable) {
            response.setError(throwable);
            logger.error("Failed to process request", throwable);
        }
        
        // Pass the response to the outbound pipeline for writing
        ctx.write(response);
        // Trigger flush to send the response immediately (can also handle flush in outbound handler)
        ctx.flush();
    }
}

Step 2: Outbound Handler for Response Writing #

This handler will intercept write operations, add success/failure listeners, and pass the message along the outbound pipeline (to encoders, then to the network).

import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RpcResponseOutboundHandler extends ChannelOutboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(RpcResponseOutboundHandler.class);

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof RpcResponse response) {
            // Add listener to track write completion
            promise.addListener(future -> {
                if (future.isSuccess()) {
                    logger.debug("Successfully sent response for request ID: {}", response.getRequestId());
                } else {
                    logger.error("Failed to send response for request ID: {}", response.getRequestId(), future.cause());
                }
            });
        }
        // Pass the message to the next outbound handler (e.g., your RpcResponseEncoder)
        super.write(ctx, msg, promise);
    }
}

Step 3: Configure the Pipeline #

Add these handlers to your channel pipeline in the correct order (remember: inbound handlers run top-to-bottom, outbound handlers run bottom-to-top):

// Inside your ServerBootstrap's childHandler configuration
ChannelPipeline pipeline = ch.pipeline();
// Add your codec handlers first (decoder for inbound, encoder for outbound)
pipeline.addLast(new RpcRequestDecoder());
pipeline.addLast(new RpcResponseEncoder());
// Add custom inbound/outbound handlers
pipeline.addLast(new RpcRequestInboundHandler());
pipeline.addLast(new RpcResponseOutboundHandler());

Option 2: Use ChannelDuplexHandler (Combine Both Logics) #

If you prefer to keep related logic in one class, ChannelDuplexHandler implements both ChannelInboundHandler and ChannelOutboundHandler, letting you handle both read and write operations in a single handler.

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RpcDuplexHandler extends ChannelDuplexHandler {
    private static final Logger logger = LoggerFactory.getLogger(RpcDuplexHandler.class);

    // Handle inbound read operations (same as the dedicated inbound handler)
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof RpcRequest request) {
            logger.debug("Received request: {}", request);
            RpcResponse response = new RpcResponse(request.getRequestId());
            
            try {
                Object result = Server.handleRequest(request);
                response.setResult(result);
            } catch (Throwable throwable) {
                response.setError(throwable);
                logger.error("Failed to process request", throwable);
            }
            
            ctx.write(response);
            ctx.flush();
        } else {
            // Pass non-RpcRequest messages to the next inbound handler
            super.channelRead(ctx, msg);
        }
    }

    // Handle outbound write operations (same as the dedicated outbound handler)
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof RpcResponse response) {
            promise.addListener(future -> {
                if (future.isSuccess()) {
                    logger.debug("Successfully sent response for request ID: {}", response.getRequestId());
                } else {
                    logger.error("Failed to send response for request ID: {}", response.getRequestId(), future.cause());
                }
            });
        }
        super.write(ctx, msg, promise);
    }
}

Configure the Pipeline for Duplex Handler #

ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RpcRequestDecoder());
pipeline.addLast(new RpcResponseEncoder());
pipeline.addLast(new RpcDuplexHandler());

Key Notes #

• Pipeline Order: Always place codec handlers (decoders/encoders) before your custom handlers. For outbound operations, the message flows from your handler to the encoder, then to the network.
• ChannelPromise: This object tracks the success/failure of write operations—using it to add listeners replaces the addListener call in your original writeAndFlush chain.
• Single Responsibility: Splitting into separate handlers makes your code more maintainable, especially as your logic grows. The duplex handler is useful for tightly coupled read/write logic.

内容的提问来源于stack exchange，提问作者yunzhitan