Dubbo之telnet实现

我们可以通过telnet来访问道对应dubbo服务的信息

比如

173716_5qEu_871390.png

我们可以利用一些指令来访问。

我们知道，默认情况下，dubbo使用netty做transport。

那么dubbo是如何区分开正常业务请求和telnet请求呢？

首先来看一下netty的服务。

NettyServer在打开是会注册一些downStream和upStream的event

    public class NettyServer extends AbstractServer implements Server {
         
        private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
     
        private Map<String, Channel>  channels; // <ip:port, channel>
     
        private ServerBootstrap                 bootstrap;
     
        private org.jboss.netty.channel.Channel channel;
     
        public NettyServer(URL url, ChannelHandler handler) throws RemotingException{
            super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
        }
     
        @Override
        protected void doOpen() throws Throwable {
            NettyHelper.setNettyLoggerFactory();
            ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
            ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
            ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
            bootstrap = new ServerBootstrap(channelFactory);
             
            final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
            channels = nettyHandler.getChannels();
            // https://issues.jboss.org/browse/NETTY-365
            // https://issues.jboss.org/browse/NETTY-379
            // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
            bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
                public ChannelPipeline getPipeline() {
                    NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
                    ChannelPipeline pipeline = Channels.pipeline();
                    /*int idleTimeout = getIdleTimeout();
                    if (idleTimeout > 10000) {
                        pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
                    }*/
                    pipeline.addLast("decoder", adapter.getDecoder());
                    pipeline.addLast("encoder", adapter.getEncoder());
                    pipeline.addLast("handler", nettyHandler);
                    return pipeline;
                }
            });
            // bind
            channel = bootstrap.bind(getBindAddress());
        }

其中decoder和encoder对应加解码，这边也对应了之前调用异常时无法通过attachment传递信息Dubbo自定义异常message过长解决

那么这边的nettyHandler最终通过层层包装委托的机制其实到了真正执行的应该是

HeaderExchangeHandler

    public void received(Channel channel, Object message) throws RemotingException {
        channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
        ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
        try {
            if (message instanceof Request) {
                // handle request.
                Request request = (Request) message;
                if (request.isEvent()) {
                    handlerEvent(channel, request);
                } else {
                    if (request.isTwoWay()) {
                        Response response = handleRequest(exchangeChannel, request);
                        channel.send(response);
                    } else {
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } else if (message instanceof Response) {
                handleResponse(channel, (Response) message);
            } else if (message instanceof String) {
                if (isClientSide(channel)) {
                    Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
                    logger.error(e.getMessage(), e);
                } else {
                    String echo = handler.telnet(channel, (String) message);
                    if (echo != null && echo.length() > 0) {
                        channel.send(echo);
                    }
                }
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }

其中根据请求message的类型进行了区分，如果是request则进行正常的业务调用，如果是String则进行telnet的回复。

这边的handler类型是ExchangeHandlerAdapter及其对应的子类。

可以确认调用telnet时继续根据spi的方法来查找对应的实现

    public String telnet(Channel channel, String message) throws RemotingException {
        String prompt = channel.getUrl().getParameterAndDecoded(Constants.PROMPT_KEY, Constants.DEFAULT_PROMPT);
        boolean noprompt = message.contains("--no-prompt");
        message = message.replace("--no-prompt", "");
        StringBuilder buf = new StringBuilder();
        message = message.trim();
        String command;
        if (message.length() > 0) {
            int i = message.indexOf(' ');
            if (i > 0) {
                command = message.substring(0, i).trim();
                message = message.substring(i + 1).trim();
            } else {
                command = message;
                message = "";
            }
        } else {
            command = "";
        }
        if (command.length() > 0) {
            if (extensionLoader.hasExtension(command)) {
                try {
                    String result = extensionLoader.getExtension(command).telnet(channel, message);
                    if (result == null) {
                        return null;
                    }
                    buf.append(result);
                } catch (Throwable t) {
                    buf.append(t.getMessage());
                }
            } else {
                buf.append("Unsupported command: ");
                buf.append(command);
            }
        }
        if (buf.length() > 0) {
            buf.append("\r\n");
        }
        if (prompt != null && prompt.length() > 0 && ! noprompt) {
            buf.append(prompt);
        }
        return buf.toString();
    }

可想而知spi在dubbo服务中完全是非常大规模的使用，可以在项目中借鉴，这也是一种很典型的控制反转 详细查看filter一级的说明

dubbo源码系列之filter的前生

回到TelnetHandler的spi文件

    clear=com.alibaba.dubbo.remoting.telnet.support.command.ClearTelnetHandler
    exit=com.alibaba.dubbo.remoting.telnet.support.command.ExitTelnetHandler
    help=com.alibaba.dubbo.remoting.telnet.support.command.HelpTelnetHandler
    status=com.alibaba.dubbo.remoting.telnet.support.command.StatusTelnetHandler
    log=com.alibaba.dubbo.remoting.telnet.support.command.LogTelnetHandler
    ls=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.ListTelnetHandler
    ps=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.PortTelnetHandler
    cd=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.ChangeTelnetHandler
    pwd=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.CurrentTelnetHandler
    invoke=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.InvokeTelnetHandler
    trace=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.TraceTelnetHandler
    count=com.alibaba.dubbo.rpc.protocol.dubbo.telnet.CountTelnetHandler

很明显这些就是telnet所提供的具体的指令。

这边以Help为例

    @Activate
    @Help(parameter = "[command]", summary = "Show help.", detail = "Show help.")
    public class HelpTelnetHandler implements TelnetHandler {
         
        private final ExtensionLoader<TelnetHandler> extensionLoader = ExtensionLoader.getExtensionLoader(TelnetHandler.class);
     
        public String telnet(Channel channel, String message) {
            if (message.length() > 0) {
                if (! extensionLoader.hasExtension(message)) {
                    return "No such command " + message;
                }
                TelnetHandler handler = extensionLoader.getExtension(message);
                Help help = handler.getClass().getAnnotation(Help.class);
                StringBuilder buf = new StringBuilder();
                buf.append("Command:\r\n    ");
                buf.append(message + " " + help.parameter().replace("\r\n", " ").replace("\n", " "));
                buf.append("\r\nSummary:\r\n    ");
                buf.append(help.summary().replace("\r\n", " ").replace("\n", " "));
                buf.append("\r\nDetail:\r\n    ");
                buf.append(help.detail().replace("\r\n", "    \r\n").replace("\n", "    \n"));
                return buf.toString();
            } else {
                List<List<String>> table = new ArrayList<List<String>>();
                List<TelnetHandler> handlers = extensionLoader.getActivateExtension(channel.getUrl(), "telnet");
                if (handlers != null && handlers.size() > 0) {
                    for (TelnetHandler handler : handlers) {
                        Help help = handler.getClass().getAnnotation(Help.class);
                        List<String> row = new ArrayList<String>();
                        String parameter = " " + extensionLoader.getExtensionName(handler) + " " + (help != null ? help.parameter().replace("\r\n", " ").replace("\n", " ") : "");
                        row.add(parameter.length() > 50 ? parameter.substring(0, 50) + "..." : parameter);
                        String summary = help != null ? help.summary().replace("\r\n", " ").replace("\n", " ") : "";
                        row.add(summary.length() > 50 ? summary.substring(0, 50) + "..." : summary);
                        table.add(row);
                    }
                }
                return "Please input \"help [command]\" show detail.\r\n" + TelnetUtils.toList(table);
            }
        }
     
    }

根据Help的注解，将各个telnet指令的功能以及详细方法打印。