Spring Boot中如何从@ShellMethod注解的方法流式输出文本

问题描述 #

使用Spring Boot 3.1.x进行Shell交互，希望从方法向控制台流式输出文本。尝试了如下代码：

@ShellMethod(key = "stream")
public Flux<String> streamMe(){
    List<String> cities = List.of("London", "Paris", "Rome", "Amsterdam","Boston");
    Flux<String> flux = Flux.fromIterable(cities);
    return flux.delayElements(Duration.ofSeconds(2));
}

但在Shell中调用该方法时，屏幕输出的是FluxConcatMapNoPrefetch而非预期的流式文本。

问题原因 #

Spring Shell默认会直接打印Flux对象的toString()结果，不会自动订阅并消费流式数据。

解决方法 #

方式一：手动订阅并输出 #

将方法改为void返回类型，通过Shell对象手动订阅Flux并输出内容，这种方式更灵活，可自定义输出逻辑：

import org.springframework.shell.Shell;
import org.springframework.shell.standard.ShellComponent;
import org.springframework.shell.standard.ShellMethod;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.List;

@ShellComponent
public class StreamCommand {

    private final Shell shell;

    public StreamCommand(Shell shell) {
        this.shell = shell;
    }

    @ShellMethod(key = "stream")
    public void streamMe() {
        List<String> cities = List.of("London", "Paris", "Rome", "Amsterdam", "Boston");
        Flux.fromIterable(cities)
            .delayElements(Duration.ofSeconds(2))
            .subscribe(
                city -> shell.write(city + "\n"),
                error -> shell.write("输出出错：" + error.getMessage() + "\n")
            );
    }
}

方式二：利用Spring Shell响应式处理机制 #

确保项目启用响应式支持（Spring Boot 3.1.x默认已包含），通过注解让框架自动处理Flux的订阅与输出：

import org.springframework.shell.standard.ShellComponent;
import org.springframework.shell.standard.ShellMethod;
import org.springframework.shell.standard.ShellMethodReturnValue;
import org.springframework.shell.reactive.ReactiveShellMethodReturnValueHandler;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.List;

@ShellComponent
public class StreamCommand {

    @ShellMethod(key = "stream")
    @ShellMethodReturnValue(handler = ReactiveShellMethodReturnValueHandler.class)
    public Flux<String> streamMe() {
        List<String> cities = List.of("London", "Paris", "Rome", "Amsterdam", "Boston");
        return Flux.fromIterable(cities).delayElements(Duration.ofSeconds(2));
    }
}

如果上述代码仍不生效，可在application.properties中明确开启响应式支持：

spring.shell.reactive.enabled=true

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