你可以使用Spring Cloud流Kafka活页夹。
创建两个流，一个用于消费，一个用于生产。
面向消费者

public interface InputStreamExample{
    String INPUT = "consumer-in";
    @Input(INPUT)
    MessageChannel readFromKafka();
}

为制片人

public interface ProducerStreamExample{

    String OUTPUT = "produce-out";

    @Output(OUTPUT)
    MessageChannel produceToKafka();
}

用于消费信息：

@StreamListener(value = ItemStream.INPUT)
public void processMessage(){
    /*
    code goes here
    */
}

用于生产

//here producerStreamExample is instance of ProducerStreamExample 
producerStreamExample.produceToKafka().send(/*message goes here*/);

现在使用binder配置consumer和producer集群，您可以将consumer in用于consumer集群，将producer out用于producing集群。
属性文件

spring.cloud.stream.binders.kafka-a.environment.spring.cloud.stream.kafka.binder.brokers:<consumer cluster>

# other properties for this binders

# bind kafka-a to consumer-in

spring.cloud.stream.bindings.consumer-in.binder=kafka-a  #kafka-a binding to consumer-in

# similary other properties of consumer-in, like

spring.cloud.stream.bindings.consumer-in.destination=<topic>
spring.cloud.stream.bindings.consumer-in.group=<consumer group>

# now configure cluster to produce

spring.cloud.stream.binders.kafka-b.environment.spring.cloud.stream.kafka.binder.brokers:<cluster where to produce>

spring.cloud.stream.bindings.produce-out.binder=kafka-b    #here kafka-b, binding to produce-out

# similary you can do other configuration like topic

spring.cloud.stream.bindings.produce-out.destination=<topic>

只需使用不同的 bootstrap.servers 属性。
如果您使用的是spring boot，请参阅
https://docs.spring.io/spring-boot/docs/current/reference/html/appendix-application-properties.html#spring.kafka.consumer.bootstrap-服务器
和
https://docs.spring.io/spring-boot/docs/current/reference/html/appendix-application-properties.html#spring.kafka.producer.bootstrap-服务器
如果你正在创建自己的工厂 @Bean s、 在那里设置属性。
https://docs.spring.io/spring-kafka/docs/current/reference/html/#connecting