如何在应用程序启动时使用 Spring Cloud Stream 发布消息？答案

【问题标题】：How to publish a message with Spring Cloud Stream on application Startup?如何在应用程序启动时使用 Spring Cloud Stream 发布消息？
【发布时间】：2018-06-13 23:13:24
【问题描述】：

我正在尝试在应用程序启动时使用 spring 云流向rabbitmq 发送消息。使用下面的示例代码。

public interface Barista {
    @Input
    SubscribableChannel orders();
}

启用绑定的 SpringBoot 应用

@SpringBootApplication
@EnableBinding(Barista.class)
public class DemoSpringCloudStreamApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoSpringCloudStreamApplication.class, args);
    }
}

应该只是在启动时发送消息的应用程序运行器

@Component
public class Startup implements ApplicationRunner {

    @Autowired
    private Barista barista;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Message<String> message = MessageBuilder.withPayload("test")
                                             .build();

        barista.orders().send(message);
    }

    //    @StreamListener("orders")
    //    public void handle(String message) {
    //        System.out.println(message);
    //    }
}

除非我取消注释上面注释掉的代码，否则上面的代码会产生下面的异常。

java.lang.IllegalStateException: Failed to execute ApplicationRunner
    at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:791) [spring-boot-2.0.2.RELEASE.jar:2.0.2.RELEASE]
    at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:778) [spring-boot-2.0.2.RELEASE.jar:2.0.2.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:335) [spring-boot-2.0.2.RELEASE.jar:2.0.2.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1255) [spring-boot-2.0.2.RELEASE.jar:2.0.2.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1243) [spring-boot-2.0.2.RELEASE.jar:2.0.2.RELEASE]
    at com.example.demospringcloudstream.DemoSpringCloudStreamApplication.main(DemoSpringCloudStreamApplication.java:12) [classes/:na]
Caused by: org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.orders'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=test, headers={id=4170f931-b303-dc96-152b-19d5c3421fb3, contentType=application/json, timestamp=1528930565229}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77) ~[spring-integration-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445) ~[spring-integration-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:394) ~[spring-integration-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]
    at com.example.demospringcloudstream.Startup.run(Startup.java:25) ~[classes/:na]
    at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:788) [spring-boot-2.0.2.RELEASE.jar:2.0.2.RELEASE]
    ... 5 common frames omitted
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:138) ~[spring-integration-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105) ~[spring-integration-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) ~[spring-integration-core-5.0.5.RELEASE.jar:5.0.5.RELEASE]
    ... 9 common frames omitted

我正在使用 Spring Boot 2.0.2 和 Sprig Cloud Stream 2.0.0，如下面的 pom 所示

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>demo-spring-cloud-stream</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>demo-spring-cloud-stream</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
        <spring-cloud.version>Finchley.RC2</spring-cloud.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-test-support</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

    <repositories>
        <repository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>


</project>

根据这个接受的答案https://stackoverflow.com/a/42600330/438319，“ApplicationRunner”应该可以工作。但是，我得到 org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers 触发事件的应用程序不需要监听事件。

那么如何在应用程序启动时使用 Spring Cloud Stream 触发事件而无需监听事件？

【问题讨论】：

标签： java spring spring-boot spring-cloud-stream

【解决方案1】：

它实际上会在错误消息中准确地告诉您它的含义。

Caused by: org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.orders'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=test, headers={id=4170f931-b303-dc96-152b-19d5c3421fb3, contentType=application/json, timestamp=1528930565229}]

您正在向application.orders 频道发送消息，但该频道没有订阅者。

【讨论】：

为什么发送消息的应用程序还必须订阅它发送消息的频道？我只想在启动时触发一条消息，并让其他人在不同的应用程序中接收它。
通道是本地的，因此其他应用程序要获取它，它必须与其中一个活页夹（例如，Kafka、Rabbit 等）一起工作。然后是输入/输出类型的通道，我看不到您的完整应用程序配置。无论如何，我不清楚您要完成什么，所以我建议您完成这个非常快速的教程（5-10 分钟）docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/… 然后我假设您会有一个更尖锐的问题
我已经更新了这个问题来澄清它。实际上，我正在深入研究 Spring Cloud Stream 阅读手册并尝试提到的每个功能。这就是我遇到这个问题的方式，看来我应该能够使用 Spring Cloud Stream 发布消息，而不必同时收听订阅我正在发布的频道。
你应该，但正如我所说，我只看到你发布的内容，并且基于你看到的错误是完全合理的并且是预期的。也许您可以将您的示例项目发布在 GitHub 上或其他我们可以查看并了解您真正想要完成的工作的地方。
github.com/amsabc/demo-spring-cloud-stream 有一个可以重现问题的 repo 感谢您的帮助。我认为这是一个错误。

【解决方案2】：

您正在将消息发送到输入频道。这就是为什么当您没有频道订阅者时会收到错误消息。

为了从您的应用程序发送消息，请尝试将您的通道定义为输出通道，例如：

public interface Barista {
  @Output
  MessageChannel orders();
}

如果您出于某种原因确实希望您的频道作为输入频道，则必须添加订阅者。

【讨论】：