您可以使用输出隔离器模式来防止Logstash在多个输出中的一个出现临时故障时被阻塞。默认情况下，Logstash在任何单个输出关闭时都会被阻塞。这种行为对于保证至少一次数据传输非常重要。
示例代码：

# config/pipelines.yml
- pipeline.id: intake
  config.string: |
    input { beats { port => 5044 } }
    output { pipeline { send_to => [es, http] } }
- pipeline.id: buffered-es
  queue.type: persisted
  config.string: |
    input { pipeline { address => es } }
    output { elasticsearch { } }
- pipeline.id: buffered-http
  queue.type: persisted
  config.string: |
    input { pipeline { address => http } }
    output { http { } }

字符串
我希望这对你有帮助

首先，确保Kafka brokers服务端口已启动并运行。在您的情况下，如您在请求中提到的9091。telnet kafka_server1 9091 telnet kafka_server2 9091 telnet kafka_server3 9091如果没有，则重新启动Kafka服务
然后在logstash输出中按照下一个格式设置所有代理。格式是host1：port1，host2：port2，列表可以是代理的子集或指向代理子集的VIP。

input {
  # Your input configuration, e.g., file, beats, etc.
}

filter {
  # Your filter configuration, if needed
}

output {
  kafka {
    #### List of Kafka bootstrap servers
    bootstrap_servers => "kafka_server1:9091,kafka_server2:9091,kafka_server3:9091"

    ####  Kafka topic to which Logstash will send messages
    topic_id => "your_kafka_topic"

    ####  Optional settings (you may need to adjust based on your Kafka setup)
    codec => "json"
    acks => "all"
  }
}

字符串