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

Question

我正在尝试在应用程序启动时使用spring cloud stream向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”应该可以工作。然而，我得到了组织。springframework。集成。MessageDispatchingException：Dispatcher没有订阅者触发事件的应用程序不需要侦听事件。

那么，如何在应用程序启动时使用Spring Cloud Stream触发事件，而不必侦听该事件呢？

Answer 1

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

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

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

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

Answer 2

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

原因：org.springframework.messaging.MessageDeliveryException：Dispatcher没有频道'application.orders'的订阅者。；嵌套异常是org.springframework.integration.MessageDispatchingException：Dispatcher没有订阅者，失败消息=GenericMessage[有效负载=test，标头={id=4170f931-b303-dc96-152b-19d5c3421fb3，contentType=应用程序/json，时间戳=1528930565229}]

您正在向应用程序发送消息。订购频道，但该频道没有订户。