快捷導(dǎo)航

解決SpringCloudStream整合Kafka,兩個(gè)通道對(duì)應(yīng)同一個(gè)topic報(bào)錯(cuò)的情況

更新時(shí)間：2026年05月06日 09:09:58 作者：加把勁騎士RideOn

文章指出通道需唯一對(duì)應(yīng)topic,否則會(huì)報(bào)錯(cuò),因兩個(gè)通道共用一個(gè)topic導(dǎo)致綁定失敗,通過(guò)修改配置文件,為不同通道分配不同的topic解決

報(bào)錯(cuò)日志

2022-05-25 14:46:03.697 ERROR 17108 --- [ask-scheduler-1] o.s.cloud.stream.binding.BindingService : Failed to create consumer binding; retrying in 30 seconds
。。。
org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer:
。。。
Caused by: org.springframework.beans.factory.support.BeanDefinitionOverrideException: Invalid bean definition with name 'Evad.consumer-group-evad.errors.recoverer' defined in null。。。

問(wèn)題所在

yml配置文件中定義的兩個(gè)通道：evad_input和devilvan_input，卻共用了一個(gè)topic：Evad，導(dǎo)致綁定失敗。

配置文件

spring:
  application:
    name: devilvan-kafka
  cloud:
    stream:
      default-binder: kafka
      bindings:
        evad_input:
          destination: Evad
          binder: kafka
          group: consumer-group-evad
          content-type: text/plain
        evad_output:
          destination: Evad
          binder: kafka
          content-type: text/plain
        devilvan_input:
          # 一個(gè)通道只能唯一對(duì)應(yīng)一個(gè)topic，否則會(huì)報(bào)binder
          destination: Evad
          binder: kafka
          # 一個(gè)消費(fèi)者組可以被多個(gè)通道使用
          group: consumer-group-evad
          content-type: text/plain
        devilvan_output:
          destination: Evad
          binder: kafka
          content-type: text/plain

解決方法

新定義一個(gè)Topic：Evad05，使devilvan通道對(duì)應(yīng)topic，區(qū)別于evad通道對(duì)應(yīng)的topic

修改后

spring:
  application:
    name: devilvan-kafka
  cloud:
    stream:
      default-binder: kafka
      bindings:
        evad_input:
          destination: Evad
          binder: kafka
          group: consumer-group-evad
          content-type: text/plain
        evad_output:
          destination: Evad
          binder: kafka
          content-type: text/plain
        devilvan_input:
          # 一個(gè)通道只能唯一對(duì)應(yīng)一個(gè)topic，否則會(huì)報(bào)binder
          destination: Evad05
          binder: kafka
          # 一個(gè)消費(fèi)者組可以被多個(gè)通道使用
          group: consumer-group-evad
          content-type: text/plain
        devilvan_output:
          destination: Evad05
          binder: kafka
          content-type: text/plain

代碼

1. XXXController（生產(chǎn)消息的控制器）

    @PostMapping("sendEvadMessage")
    public ResultMessage<String> sendEvadMessage(@RequestBody String message) {
        ResultMessage<String> resultMessage = new ResultMessage<>();
        sender.sendEvadMessage(message);
        resultMessage.setData(message);
        return resultMessage.success();
    }

    @PostMapping("sendDevilvanMessage")
    public ResultMessage<String> sendDevilvanMessage(@RequestBody String message) {
        ResultMessage<String> resultMessage = new ResultMessage<>();
        sender.sendDevilvanMessage(message);
        resultMessage.setData(message);
        return resultMessage.success();
    }

2. 自定義通道

public interface EvadChannel {
    String EVAD_INPUT = "evad_input";
    String EVAD_OUTPUT = "evad_output";
    String DEVILVAN_INPUT = "devilvan_input";
    String DEVILVAN_OUTPUT = "devilvan_output";

    /**
     * 缺省接收消息通道
     * @return channel 返回缺省信息接收通道
     */
    @Input(EVAD_INPUT)
    MessageChannel receiveEvadMessage();

    /**
     * 缺省發(fā)送消息通道
     * @return channel 返回缺省信息發(fā)送通道
     */
    @Output(EVAD_OUTPUT)
    MessageChannel sendEvadMessage();

    /**
     * 缺省接收消息通道
     * @return channel 返回缺省信息接收通道
     */
    @Input(DEVILVAN_INPUT)
    MessageChannel receiveDevilvanMessage();

    /**
     * 缺省發(fā)送消息通道
     * @return channel 返回缺省信息發(fā)送通道
     */
    @Output(DEVILVAN_OUTPUT)
    MessageChannel sendDevilvanMessage();
}

3. EvadMessageSender（通過(guò)通道發(fā)送消息）

@Slf4j
@Component
public class EvadMessageSender {
    @Autowired
    private EvadChannel channel;

    /**
     * 消息發(fā)送到默認(rèn)通道：缺省通道對(duì)應(yīng)缺省主題
     *
     * @param message
     */
    public void sendEvadMessage(String message) {
        channel.sendEvadMessage().send(MessageBuilder.withPayload(message).build());
    }

    /**
     * 消息發(fā)送到默認(rèn)通道：缺省通道對(duì)應(yīng)缺省主題
     *
     * @param message
     */
    public void sendDevilvanMessage(String message) {
        channel.sendDevilvanMessage().send(MessageBuilder.withPayload(message).build());
    }
}

4. EvadReceiveListener（訂閱/消費(fèi)者）

@Slf4j
@Configuration
@EnableBinding(value = EvadChannel.class)
public class EvadReceiveListener {
    @StreamListener(EvadChannel.EVAD_INPUT)
    public void receiveEvadMessage(Message<String> message) {
        log.info("{}    訂閱消息：通道 = " + EvadChannel.EVAD_INPUT + "，data = {}",
                DateUtil.now(), message.getPayload());
    }

    @StreamListener(EvadChannel.DEVILVAN_INPUT)
    public void receiveDevilvanMessage(Message<String> message) {
        log.info("{}    訂閱消息：通道 = " + EvadChannel.DEVILVAN_INPUT + "，data = {}",
                DateUtil.now(), message.getPayload());
    }
}