Spring Cloud Stream - 如何避免向非目标代理发送消息答案

【问题标题】：Spring Cloud Stream - How to avoid sending messages to non-target brokersSpring Cloud Stream - 如何避免向非目标代理发送消息
【发布时间】：2022-10-25 15:07:34
【问题描述】：

我有一个具有以下配置的 kafka 生产者应用程序。

spring:
  main:
    banner-mode: off
  application:
    name: sample-app

  cloud:
    stream:
      function:
        definition: sendEvents;jsonEvents

      binders:
        kafka1:
          type: kafka
          environment:
            spring.cloud.stream.kafka.binder:
              brokers:
                - 'localhost:29092'

        kafka2:
          type: kafka
          environment:
            spring.cloud.stream.kafka.binder:
              brokers:
                - 'localhost:29093'

      bindings:
        sendEvents-out-0:
          binder: kafka1
          destination: send_events
          contentType: application/json
        jsonEvents-out-0:
          binder: kafka2
          destination: json_events
          contentType: application/json
        consumer-in-0:
          binder: kafka1
          group: ${spring.application.name}
          destination: send_events
        consumer2-in-0:
          binder: kafka2
          group: ${spring.application.name}
          destination: json_events

import java.util.HashMap;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.http.MediaType;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

@RestController
@RequestMapping(path = "/{datasource}/{topicName}")
@Slf4j
public class MainController
{

  @Autowired
  StreamBridge streamBridge;

  @PostMapping(consumes = MediaType.APPLICATION_JSON_VALUE)
  public Map postMessage(@PathVariable(name = "datasource") String source,
                         @PathVariable(name = "topicName") String topic,
                         @RequestBody Object body)
  {

    Map<String, String> map = new HashMap<>();
    log.info("what is source " + source);

    log.info("what is topic " + topic);
    map.put("source", source);
    map.put("topic", topic);

    Message<Map<String, String>> message = MessageBuilder.withPayload(map)
      .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
      .setHeader(MessageHeaders.ERROR_CHANNEL, "error_channel")
      .build();

    log.info("postMessage");
    if (source.equalsIgnoreCase("send")) {
      streamBridge.send("sendEvents-out-0", message);
    } else {
      
      streamBridge.send("jsonEvents-out-0", message);
    }
    return map;


  }
}

该应用程序是一个简单的spring-boot-starter-webflux 应用程序。当收到请求时，它会将正文发送到相应的 Kafka 代理。一个代理中的消息不应该在另一个代理中。 StreamBridge 用于将消息发送到不同的绑定。

但是，当我测试应用程序时，我发现应该在 kafka2 的 (jsonEvents-out-0) 中的消息也可以在同一主题 (json_events) 的 kafka1 中找到。我怎样才能完全避免 kafka1 存储应该只在 kafka2 中的消息？

【问题讨论】：

标签： spring-cloud-stream spring-cloud-stream-binder-kafka

【解决方案1】：

我找到了解决方案。我可以将自动装配的 BindingServiceProperty 绑定到 bean 并通过 BindingServiceProperty 获取 BinderName

bindingName = "sendEvents-out-0";

String binder = this.bindingServiceProperties.getBindingProperties(bindingName).getBinder();

streamBridge.send(bindingName, binder, message);

在这种情况下，消息将不会发送到默认的 kafka 集群。

【讨论】：