Spring Cloud Stream Kafka：org.springframework.integration.MessageDispatchingException：调度程序没有订阅者答案

【问题标题】：Spring Cloud Stream Kafka: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribersSpring Cloud Stream Kafka：org.springframework.integration.MessageDispatchingException：调度程序没有订阅者
【发布时间】：2017-04-10 05:54:50
【问题描述】：

我正在尝试使用 Spring Cloud Stream Kafka Binding 通过本地 zookeeper 和 kafka 服务器发送和接收消息。但是，在启动 Spring MVC 服务器时，我看到以下异常：

Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:154) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
... 28 common frames omitted

服务端很简单，一个类一个spring应用属性文件：

package com.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.util.Map;

@RestController
@EnableBinding(ProducerChannels.class)
@SpringBootApplication
public class ProducerApplication {

  private final MessageChannel consumer;

  public ProducerApplication(ProducerChannels channels){
      this.consumer = channels.consumer();
  }

  @PostMapping("/greet/{name}")
  public void publish(@RequestBody Map<String, String> name){
      String greeting = "Hello, " + name + "!";

      Message<String> msg = MessageBuilder.withPayload(greeting).build();

      consumer.send(msg);
  }

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

interface ProducerChannels {

  @Output
  MessageChannel consumer();
}

块引用 spring.cloud.stream.kafka.bindings.consumer.destination = 消费者 server.port=8080

我还需要做什么来配置 Spring Cloud Stream？

【问题讨论】：

标签： spring apache-kafka spring-cloud-stream

【解决方案1】：

您使用的是什么版本？我刚刚将您的代码粘贴到启动 1.4.2 应用程序（流启动器为 1.1.0.RELEASE），它对我来说很好。

【讨论】：

【解决方案2】：

consumer 是一个特殊的关键字。你不应该使用它。我认为如果您按以下方式修复它，它会发生。

public interface ProducerChannels {
    
    String OUTPUT = "example-topic";

    @Output(OUTPUT)
    MessageChannel exampleTopic();
}

application.yml 文件

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      bindings:
        example-topic:
          destination: example-topic
          contentType: application/json

或者

spring.cloud.stream.bindings.exampleTopic.destination= example-topic

【讨论】：