前面一篇，將瞭如何配置kafak：https://www.cnblogs.com/hanjun0612/p/13398119.html

這一篇在上面的基礎，擴充套件成2個微服務。

其實很簡單。

就是在原來的基礎上，把StreamReceiver 和 TestStream 拷貝到Service1服務中。

程式碼如下：

StreamReceiver

package com.test.service1.controller.kafka;
 
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
 
import org.springframework.stereotype.Component;
 
/**
 * @author Tyler
 * @date 2020/7/28
 */
 
@Component
@EnableBinding(value = {TestStream.class})
public class StreamReceiver {
 
    @StreamListener(TestStream.INPUT)
    public void receive(String message) {
        System.out.println("StreamReceiver: "+message);
    }
}
 
 

TestStream

package com.test.service1.controller.kafka;
 
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
 
/**
 * @author
 
 Tyler
 * @date 2020/7/28
 */
 
public interface TestStream {
    String INPUT = "test-in";
    String OUTPUT = "test-out";
    @Input(INPUT)
    SubscribableChannel testIn();
    @Output(OUTPUT)
    MessageChannel testOut();
 
}

POM檔案：

PS：這裡spring版本是2.3.2，用的是Hoxton.SR6

之前用的1.5.4，會報錯：kafka Could not convert message

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    
    <properties>
        <java.version>1.8</java.version>
        <spring-cloud.version>Hoxton.SR6</spring-cloud.version>
    </properties>
 
 
<dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
            <version>2.0.1.RELEASE</version>
        </dependency>
</dependencies>

application.yml

spring:
  application:
    name: service1
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      bindings:
        test-in: #TestStream 中 INPUT
          destination: testkafka

效果

Service1:

Kafka: