当前位置：首页 > news >正文

【Netty】消息分发处理方式

news 来源：原创 2025/5/9 20:04:27

背景

检索平台，使用长链接的方式与外部系统进行交互; 因平台使用的的自定义二进制交互协议，因此需要针对每个接口请求与响应都要进行编解码, 因此需要一种针对不同消息的分发处理

方案一注解 + 反射

示例：

/**
* 消息类型注解
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface MessageHandler {
    //类型,平台使用符号CMD, 为了兼容不同类型,相同处理方式逻辑，这里可设置为数组
    String type();
    // 优先级
    int priority() default 0;
}

/**
* 基础处理器
*/
public abstract class AbstractMessageHandler {
    // 处理消息的抽象方法
    public abstract void handle(Object message);
}


/**
* 具体处理器示例
*/
@MessageHandler(type = "USER")
public class UserMessageHandler extends AbstractMessageHandler {
    @Override
    public void handle(Object message) {
        // 处理用户消息
    }
}

@MessageHandler(type = "ORDER")
public class OrderMessageHandler extends AbstractMessageHandler {
    @Override
    public void handle(Object message) {
        // 处理订单消息
    }
}

/**
  * 消息分发处理器
  */
public class MessageDispatchHandler extends ChannelInboundHandlerAdapter {
    private final Map<String, AbstractMessageHandler> handlerMap = new HashMap<>();

    
    public MessageDispatchHandler() {
        // 初始化时扫描并注册所有处理器
        initHandlers();
    }

    private void initHandlers() {
        // 扫描指定包下的所有类
        Reflections reflections = new Reflections("com.your.package");
        Set<Class<?>> handlers = reflections.getTypesAnnotatedWith(MessageHandler.class);
        
        for (Class<?> handlerClass : handlers) {
            MessageHandler annotation = handlerClass.getAnnotation(MessageHandler.class);
            try {
                AbstractMessageHandler handler = (AbstractMessageHandler) handlerClass.newInstance();
                handlerMap.put(annotation.type(), handler);
            } catch (Exception e) {
                // 异常处理
            }
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 获取消息类型
        String type = getMessageType(msg);
        AbstractMessageHandler handler = handlerMap.get(type);
        
        if (handler != null) {
            handler.handle(msg);
        } else {
            // 处理未知类型消息
            ctx.fireChannelRead(msg);
        }
    }
}


/**
* 异常处理
*/
public class MessageDispatchHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 统一的异常处理
        if (cause instanceof MessageHandleException) {
            // 处理特定异常
        } else {
            // 处理其他异常
        }
    }
}


/**
* 处理器生命周期管理
*/
public abstract class AbstractMessageHandler {
    public void init() {
        // 初始化逻辑
    }

    public void destroy() {
        // 销毁逻辑
    }

    public abstract void handle(Object message);
}

优势：

更容易实现动态的消息类型处理
可以更方便地进行统一的消息处理前后的拦截
便于实现消息处理的统计、监控等横切功能

问题：

反射可能带来性能开销
类型判断的逻辑集中在一处，可能会变得复杂
不是标准的 Netty Pipeline 模式，不能像 Netty Pipeline 那样灵活地调整处理顺序

方案二、Netty责任链

// 配置 Pipeline
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new StringDecoder());
    pipeline.addLast(new UserMessageHandler());
    pipeline.addLast(new OrderMessageHandler());
    pipeline.addLast(new SystemMessageHandler());
}

// 各个 Handler 只处理自己关心的消息类型
public class UserMessageHandler extends SimpleChannelInboundHandler<UserMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, UserMessage msg) {
        // 处理用户消息
    }
}

public class OrderMessageHandler extends SimpleChannelInboundHandler<OrderMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, OrderMessage msg) {
        // 处理订单消息
    }
}