SpringCloudStream同时支持rabbitmq与kafka消息收发

说明

以下的方案仅仅是针对阿里云的kafka，因为我们阿里云的kafka没暴露zookeeper，无法通过spring cloud stream binder的方式连接，如果需要通过这种方式，zookeeper要确保是能连上的，否则只能通过以下的方式。
附： 这是官方文档 https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.2.RELEASE/reference/html/index.html

一 application.yaml的配置如下

spring:
  rabbitmq:
    host: rabbitmq-server
    port: 5672
    username: admin
    password: 123456
    virtual-host: /cloud
  kafka:
    bootstrap-servers: 172.24.30.35:9092,172.24.30.36:9092,172.24.30.38:9092
    listener:
      concurrency: 3
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: bdloader-service
      enable-auto-commit: true
      auto-offset-reset: earliest
      max-poll-records: 1
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  cloud:
    stream:
      bindings:
        xxx_input01:
          destination: xxx_events
          group: xxx-service
        xxx_output01:
          destination: xxx_events

二 rabbitmq与kafka的maven依赖如下

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <!-- 该版本根据你所使用的spring boot版本而定 不同版本不一样 具体参考官方文档 [https://spring.io/projects/spring-kafka](https://spring.io/projects/spring-kafka) -->
    <version>1.1.1.RELEASE</version>
</dependency>

三 Java代码

rabbitmq示例如下：

public interface XxxChannel {
    String INPUT = "xxx_input01";
    String OUTPUT = "xxx_output01";

    @Input(XxxChannel.INPUT)
    SubscribableChannel input();
    @Output(OUTPUT)
    MessageChannel output();
}

@EnableBinding(XxxChannel.class)
public class XxxConsumer {
    @StreamListener(value = XxxChannel.INPUT)
    public void handle(String msg) {
        ...
    }
}

@EnableBinding(XxxChannel.class)
public class XxxPublisher {
    @Autowired
    private XxxChannel xxxChannel;

    /**
     * 发送到mq
     * @param msg
     */
    public void send(String msg) {
        //发送到mq
        xxxChannel.output().send(MessageBuilder.withPayload(msg).build());
    }
}

kafka示例如下：

// 收
@KafkaListener(topics = "kafka_xxx_envents")
public void receive(String msg) {
    // kafka消息处理逻辑
}
 
// 发
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void send(String msg) {
    kafkaTemplate.send("kafka_xxx_envents", msg);
}