如何在Project Reactor中集成sshd mina/netty实现响应式SSH

现有代码拿不到命令返回的核心问题 #

• 测试用例没等异步流跑完就退出：JUnit默认主线程执行完就终止JVM，你调用subscribe是非阻塞的，写命令完成后的读回调还没来得及执行，进程就已经关了，自然看不到输出。
• 阻塞调用堵死响应式线程：connect().verify()、auth().verify()、channel.open().verify()都是阻塞方法，直接跑在响应式流水线里会占住WebFlux的NIO事件循环，导致IO事件没法被处理。
• 读响应逻辑设计有缺陷：每次调用readResponse都新建独立的缓冲区和监听器，平时不会持续消费通道里的数据，只有发完命令才临时注册读监听，很容易漏掉已经到达缓冲区的响应；另外靠匹配$ 判断响应结束的逻辑太脆，不管是root用户的#提示符、带自定义路径/颜色的提示符、提示符后没空格的情况，都会匹配失败导致无限等待。
• 通道选型不对：createShellChannel是给交互式终端用的，输出里混了命令回显、终端控制字符、提示符，本来就不适合串行跑命令拿结果的场景。

适配Project Reactor的正确实现 #

先修测试 #

测试异步流必须加阻塞等待，或者用reactor-test的StepVerifier，最简单的改法：

@Test
void run() throws Exception {
    try (SshDocker sshClient = new SshDocker()) {
        // 阻塞等连接完成
        String welcomeMsg = sshClient.open().block(Duration.ofSeconds(10));
        System.out.println(welcomeMsg);
        // 阻塞等命令执行完成
        String lsResult = sshClient.execCommand("ls\n", Duration.ofSeconds(3)).block(Duration.ofSeconds(5));
        System.out.println(lsResult);
    }
}

调整线程模型 #

所有SSHD的阻塞操作，必须通过subscribeOn(Schedulers.boundedElastic())调度到专门的阻塞线程池，绝对不能直接跑在Reactor的NIO事件循环线程上。

优化通道与读逻辑 #

如果是串行执行命令拿返回，优先用createExecChannel而不是Shell通道，每个命令单独开exec通道，命令跑完通道自动关闭，直接读到EOF就是完整返回，不需要靠提示符判断结束，逻辑简单很多。如果必须用交互式Shell（比如要切sudo、跑多步交互命令），就把通道的异步输出包装成一个持续消费的热Flux，后台一直读数据，发命令的时候再从输出流里拆分对应命令的结果，不要每次发命令才临时注册读监听。
另外你用的2.8.0版本sshd-netty有已知的异步读EOF不触发回调的bug，建议升级到2.10.x以上的稳定版本。
核心的Exec通道适配代码片段：

public Mono<String> execCommand(String command, Duration timeout) {
    return Mono.<String>create(sink -> {
        // 每个命令独立创建exec通道，自动处理资源关闭
        try (ClientChannel execChannel = session.createExecChannel(command)) {
            execChannel.setStreaming(StreamingChannel.Streaming.Async);
            ByteArrayOutputStream respOut = new ByteArrayOutputStream();
            execChannel.setOut(new NoCloseOutputStream(respOut));
            execChannel.open().verify(timeout);

            IoInputStream asyncOut = execChannel.getAsyncOut();
            Buffer readBuffer = new ByteArrayBuffer();
            // 递归读直到EOF
            asyncOut.read(readBuffer).addListener(new SshFutureListener<IoReadFuture>() {
                @Override
                public void operationComplete(IoReadFuture future) {
                    try {
                        future.verify(timeout);
                        int available = readBuffer.available();
                        if (available > 0) {
                            respOut.write(readBuffer.array(), readBuffer.rpos(), available);
                            readBuffer.rpos(readBuffer.rpos() + available);
                            readBuffer.compact();
                            asyncOut.read(readBuffer).addListener(this);
                        } else {
                            // 读到EOF，返回结果
                            sink.success(respOut.toString(StandardCharsets.UTF_8));
                        }
                    } catch (Exception e) {
                        sink.error(e);
                    }
                }
            });
        } catch (Exception e) {
            sink.error(e);
        }
    })
    .subscribeOn(Schedulers.boundedElastic())
    .timeout(timeout);
}

替代方案 #

如果觉得自己适配异步SSHD的逻辑太容易出问题，可以选更简单的方案：

• 直接用SSHD或者JSch的阻塞客户端，把整个命令执行逻辑包在Mono.fromCallable()里，调度到Schedulers.boundedElastic()线程池运行，开发成本极低，QPS不高的场景性能完全够用，不用自己处理复杂的异步回调适配。
• 直接基于Reactor Netty做SSH协议适配，把Netty的IO事件流直接转成Reactor的Flux，减少中间适配层的坑。
• 选已经封装好的响应式SSH客户端，直接提供Mono/Flux风格的API，不需要自己处理Future、回调、资源释放的逻辑。

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