在 kafka 主题中创建多个分区并使用 kafka-node 向所有分区发布消息

Question

我是 kafka 的新手，使用 kafka-node 在 nodeJS 中实现它。我想在一个主题中创建 3 个分区，并同时向所有主题发布消息。我尝试了以下代码，但这里只有一个分区正在创建，所有消息都将发送到该分区。谁能告诉我哪里出错了。非常感谢。

Abc.abcData = async() => {
    try 
    {
        var client = new kafka.KafkaClient();
        var topic = 'newTopic';
        var topicsToCreate = [
        {
            topic: topic,
            partitions: 3,
            replicationFactor: 2,
            replicaAssignment: [
            {
              partition: 0,
              replicas: [0]
            },
            {
                partition: 1,
                replicas: [1]
            },
            {
                partition: 2,
                replicas: [2]
            }
          ]
        },
        ]
        client.createTopics(topicsToCreate, (error, result) => {
        console.log(result);
        });
        var HighLevelProducer = kafka.HighLevelProducer;
        var producer = new HighLevelProducer(client);
        var payloads = [
        { topic: topic, messages: 'this is partition 1!!', partitions: 0},
        { topic: topic, messages: 'this is partition 2!!', partitions: 1},
        { topic: topic, messages: 'this is partition 3!!', partitions: 2}
        ];
        producer.on('ready', function () {
            producer.send(payloads, function (err, result) {
                if (err)
                    console.log(err);
                console.log(result);

            });
        });
        
    } 
    catch (err) 
    {
        console.error(err.message);
    }
};

我收到如下回复 -

[ { topic: 'newTopic', error: "Topic 'newTopic' already exists." } ]
{"newTopic":{"0":6}}

Answer 1

这里你在kafka服务器上使用了createTopics()，它只在Kafka服务器上auto.create.topics.enable时有效, 设置为真。客户端只需向将自动创建主题的服务器发送元数据请求。当 async 设置为 false 时，此方法不会 return 直到创建所有主题，否则它会立即 return。所以，这里默认创建一个topic，一个partition。要创建多个分区或使其自定义，您必须在 server.property 文件中添加以下行 -

auto.create.topics.enable=false

在 kafka 主题中创建多个分区并使用 kafka-node 向所有分区发布消息

Create multiple partitions in kafka topic and publish message to all of them using kafka-node

node.js

apache-kafka

kafka-node