<!-- kafka -->
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

spring: 
 cloud:
  stream:
   kafka:
    binder:
     brokers: xxx.xxx.xxx.xx:xxxx // Kafka的消息中間件服務(wù)器地址
   bindings:
    xxx_output: // 通道名稱 
     destination: xxx // 消息發(fā)往的目的地，對應(yīng)topic 在發(fā)送消息的配置里面，group是不用配置的 
     // 如果我們需要傳輸json的信息，那么在發(fā)送消息端需要設(shè)置content-type為json（其實可以不寫，默認(rèn)content-type就是json）
    xxx_input:
     destination: xxx // 消息發(fā)往的目的地，對應(yīng)topic
     group: xxx // 對應(yīng)kafka的group

@EnableBinding(Source.class) // @EnableBinding 是綁定通道的，Soure.class是spring 提供的，表示這是一個可綁定的發(fā)布通道
@Service
public class MqService {

  @Resource(name = KafkaConstants.OES_WORKBENCH_LIFE_DATA_OUTPUT)
  private MessageChannel oesWorkbenchChannel;

  /**
   * 發(fā)送一條kafka消息
   */
  public boolean sendLifeData(Object object) {
    return MqUtils.send(oesWorkbenchChannel, object, KafkaConstants.OES_WORKBENCH_LIFE_DATA_OUTPUT);
  }
}

// 發(fā)布通道
public interface Source {
  @Output(KafkaConstants.OES_WORKBENCH_LIFE_DATA_OUTPUT)
  MessageChannel oesWorkbenchLifeDataOutput(); // 發(fā)布通道用MessageChannel 
}

@Slf4j
@EnableBinding(Sink.class)
public class WorkbenchStreamListener {

  @Resource
  private FileService fileService;

  @StreamListener(KafkaConstants.xxx_input) // 監(jiān)聽接受通道
  public void receiveData(MoveMessage moveMessage) {
  }
}

// 接受通道
public interface Sink {
  @Input(KafkaConstants.OES_WORKBENCH_MOVE_INPUT)
  SubscribableChannel oesWorkbenchMoveInput(); // 接受通道用SubscribableChannel 
}