java - Kafka 生产者配置 metadata.broker.list 和 url

Question

我已经设置了一个有 3 个代理的 kafka 服务器。我想从我的计算机向这三个代理发送消息，但我已经为每个代理配置了一个 url，例如 abc.com/kafka1/abc.com/kafka2/abc.com/kafka3/ ngix.

如何在 metadata.broker.list 属性中使用这些 url？我的代码如下。

package com.xxx.x.kafka.producer;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Date;
import java.util.Properties;
import java.util.Random;


    class TestProducer {
        public static void main(String[] args) {
            long events = Long.parseLong(args[0]);
            Random rnd = new Random();

            Properties props = new Properties();
            props.put("metadata.broker.list", "abc.com/kafka1/:80,abc.com/kafka2/:80,abc.com/kafka3/:80");

            props.put("serializer.class", "kafka.serializer.StringEncoder");
            props.put("partitioner.class", "com.knx.adx.kafka.producer.SimplePartitioner");
            props.put("request.required.acks", "1");

            ProducerConfig config = new ProducerConfig(props);

            Producer<String, String> producer = new Producer<String, String>(config);

            for (long nEnvents = 0; nEnvents < events; nEnvents++) {
                long runtime = new Date().getTime();
                String ip = "192.168.2." + rnd.nextInt(255);
                String msg = runtime + ",www.example.com" + ip;
                KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
                producer.send(data);
            }
            producer.close();
        }
    }

这是我在运行代码时遇到的错误。

Exception in thread "main" kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
    at kafka.producer.Producer.send(Producer.scala:77)
    at kafka.javaapi.producer.Producer.send(Producer.scala:33)
    at com.knx.adx.kafka.producer.TestProducer.main(TestProducer.java:35)

line error producer.send(data);

Answer 1

配置变量 metadata.broker.list 需要 host1:port1,host2:port2 而不是 URL。尝试为每个代理配置不同的子域名，例如 kafka1.abc.com:80,kafka2.abc.com:80,kafka3.abc.com:80 并将这些子域指向适当的主机。请参阅 Kafka Configuration 的生产者配置部分

This is for bootstrapping and the producer will only use it for getting metadata (topics, partitions and replicas). The socket connections for sending the actual data will be established based on the broker information returned in the metadata. The format is host1:port1,host2:port2, and the list can be a subset of brokers or a VIP pointing to a subset of brokers.