Spring Cloud Stream Kotlin 消费者问题答案

【问题标题】：Spring Cloud Stream Kotlin Consumer ProblemSpring Cloud Stream Kotlin 消费者问题
【发布时间】：2020-05-27 03:34:52
【问题描述】：

我正在尝试使用 Kotlin 的 Spring Cloud Stream。我写了一个简单的消费者如下：

@Bean
fun log(): Consumer<Person> {
    return Consumer<Person> { person: Person -> println("Received: ${person.name}") }
}

Spring Cloud Stream 没有找到这个消费者并立即退出。它不会给出错误，只是终止。

如果我在 Java 中编写与以下内容完全相同的消费者，Stream Cloud Stream 会找到该消费者并使用输入。

@Bean
public Consumer<Person> log() {
    return person -> {
        System.out.println("Received: " + person.getName());
    };
}

为什么 Spring Cloud Stream 在 Kotlin 中找不到我的使用者？

这是 Spring Boot 控制台输出：

2020-05-26 08:01:29.167  INFO 78463 --- [           main] c.a.s.t.TestBinderApplicationKt          : Starting TestBinderApplicationKt on mburbidg-macOS with PID 78463 (/Users/mburbidg/test-binder/build/classes/kotlin/main started by mburbidg in /Users/mburbidg/test-binder)
2020-05-26 08:01:29.181  INFO 78463 --- [           main] c.a.s.t.TestBinderApplicationKt          : No active profile set, falling back to default profiles: default
2020-05-26 08:01:30.109  INFO 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
2020-05-26 08:01:30.113  INFO 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'taskScheduler' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.
2020-05-26 08:01:30.114 DEBUG 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : The '#jsonPath' SpEL function cannot be registered: there is no jayway json-path.jar on the classpath.
2020-05-26 08:01:30.114 DEBUG 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : SpEL function '#xpath' isn't registered: there is no spring-integration-xml.jar on the classpath.
2020-05-26 08:01:30.116  INFO 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
2020-05-26 08:01:30.145  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.147  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration$IntegrationJmxConfiguration' of type [org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration$IntegrationJmxConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.153  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration' of type [org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.156  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'mbeanServer' of type [com.sun.jmx.mbeanserver.JmxMBeanServer] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.166  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.167  INFO 78463 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationDisposableAutoCreatedBeans' of type [org.springframework.integration.config.annotation.Disposables] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-05-26 08:01:30.468  INFO 78463 --- [           main] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService 'taskScheduler'
2020-05-26 08:01:30.484  INFO 78463 --- [           main] c.f.c.c.BeanFactoryAwareFunctionRegistry : Looking up function '' with acceptedOutputTypes: []
2020-05-26 08:01:30.601 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering beans for JMX exposure on startup
2020-05-26 08:01:30.602  INFO 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel errorChannel
2020-05-26 08:01:30.603 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Located managed bean 'org.springframework.integration:type=MessageChannel,name=errorChannel': registering with JMX server as MBean [org.springframework.integration:type=MessageChannel,name=errorChannel]
2020-05-26 08:01:30.660  INFO 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel nullChannel
2020-05-26 08:01:30.660 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Located managed bean 'org.springframework.integration:type=MessageChannel,name=nullChannel': registering with JMX server as MBean [org.springframework.integration:type=MessageChannel,name=nullChannel]
2020-05-26 08:01:30.678  INFO 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageHandler errorLogger
2020-05-26 08:01:30.679 DEBUG 78463 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Located managed bean 'org.springframework.integration:type=MessageHandler,name=errorLogger,bean=internal': registering with JMX server as MBean [org.springframework.integration:type=MessageHandler,name=errorLogger,bean=internal]
2020-05-26 08:01:30.700 DEBUG 78463 --- [           main] faultConfiguringBeanFactoryPostProcessor : 
Spring Integration global properties:

spring.integration.channels.autoCreate=true
spring.integration.taskScheduler.poolSize=10
spring.integration.readOnly.headers=
spring.integration.channels.maxUnicastSubscribers=0x7fffffff
spring.integration.endpoints.noAutoStartup=
spring.integration.messagingTemplate.throwExceptionOnLateReply=false
spring.integration.channels.maxBroadcastSubscribers=0x7fffffff

2020-05-26 08:01:30.701 DEBUG 78463 --- [           main] .s.i.c.GlobalChannelInterceptorProcessor : No global channel interceptors.
2020-05-26 08:01:30.703  INFO 78463 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2020-05-26 08:01:30.703  INFO 78463 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 1 subscriber(s).
2020-05-26 08:01:30.703  INFO 78463 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
2020-05-26 08:01:30.716  INFO 78463 --- [           main] c.a.s.t.TestBinderApplicationKt          : Started TestBinderApplicationKt in 2.32 seconds (JVM running for 3.12)
2020-05-26 08:01:30.722  INFO 78463 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : Removing {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2020-05-26 08:01:30.723  INFO 78463 --- [extShutdownHook] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 0 subscriber(s).
2020-05-26 08:01:30.723  INFO 78463 --- [extShutdownHook] o.s.i.endpoint.EventDrivenConsumer       : stopped bean '_org.springframework.integration.errorLogger'
2020-05-26 08:01:30.724  INFO 78463 --- [extShutdownHook] o.s.s.c.ThreadPoolTaskScheduler          : Shutting down ExecutorService 'taskScheduler'
2020-05-26 08:01:30.726 DEBUG 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Unregistering JMX-exposed beans on shutdown
2020-05-26 08:01:30.726 DEBUG 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Unregistering JMX-exposed beans
2020-05-26 08:01:30.727  INFO 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Summary on shutdown: bean 'errorChannel'
2020-05-26 08:01:30.727  INFO 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Summary on shutdown: nullChannel
2020-05-26 08:01:30.727  INFO 78463 --- [extShutdownHook] o.s.i.monitor.IntegrationMBeanExporter   : Summary on shutdown: bean '_org.springframework.integration.errorLogger.handler' for component '_org.springframework.integration.errorLogger'

Process finished with exit code 0

【问题讨论】：

标签： kotlin spring-cloud-stream consumer

【解决方案1】：

我只是用这个测试过它；并且没有问题...

@SpringBootApplication
open class So62025232Application {
    @Bean
    open fun input(): Consumer<String> {
        return Consumer { x: String? -> println(x) }
    }

    @Bean
    open fun runner(template: RabbitTemplate): ApplicationRunner {
        return ApplicationRunner { args: ApplicationArguments? ->
            template.convertAndSend("input-in-0.foo", "foo")
            Thread.sleep(2000)
        }
    }

    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            val ctx = SpringApplication.run(So62025232Application::class.java, *args)
            println(ctx.getBeansOfType(Consumer::class.java))
            ctx.close()
        }
    }
}

spring.cloud.stream.bindings.input-in-0.group=foo

foo
{input=com.example.demo.So62025232Application$input$1@2a65bb85}

Java 版本

@SpringBootApplication
public class So62025232Application {

    public static void main(String[] args) {
        ConfigurableApplicationContext ctx = SpringApplication.run(So62025232Application.class, args);
        System.out.println(ctx.getBeansOfType(Consumer.class));
        ctx.close();
    }

    @Bean
    Consumer<String> input() {
        return System.out::println;
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> {
            template.convertAndSend("input-in-0.foo", "foo");
            Thread.sleep(2000);
        };
    }

}

【讨论】：

【解决方案2】：

谢谢！您的回答帮助我找出了问题所在。我的应用确实是这样的：

@SpringBootApplication
open class TestBinderApplication

fun main(args: Array<String>) {
    runApplication<TestBinderApplication>(*args)
}

@Bean
fun log(): Consumer<Person> {
    return Consumer<Person> { person: Person -> println("Received: ${person.name}") }

当我把它改成：

@SpringBootApplication
open class TestBinderApplication {
    @Bean
    open fun log(): Consumer<Person> {
        return Consumer<Person> { person: Person -> println("Received: ${person.name}") }
    }

    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            runApplication<TestBinderApplication>(*args)
        }
    }
}

成功了！

我是 Kotlin 的新手，所以我不确定有什么区别。

【讨论】：