1

您对 Sarama 和 Kafka 使用什么配置值？

卡夫卡版本：kafka_2.12-1.1.0.tgz Go 版本：1.9.1

package kafka

import (
    "flag"
    "fmt"
    "log"
    "strings"

    "github.com/Shopify/sarama"
)

var partition = flag.Int("partition", 12, "The partition to produce to.")

func Start_producer(payload []byte) {
    flag.Parse()
    s := "mydata"
    topic := &s
    // brokers := &[]string{"172.25.33.175:9092,172.25.33.176:9092,172.25.33.177:9092"}
    // brokers := []string{"172.25.33.175:9092,172.25.33.176:9092,172.25.33.177:9092"}

    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Retry.Max = 5
    config.Producer.Return.Successes = true
    producer, err := sarama.NewSyncProducer(strings.Split("172.25.33.175:9092,172.25.33.176:9092,172.25.33.177:9092", ","), config) //default port

    if err != nil {
        log.Println("ERRR")
        panic(err)
    }
    defer func() {
        if err := producer.Close(); err != nil {
            panic(err)
        }
    }()
    msg := &sarama.ProducerMessage{
        Topic:     *topic,
        Value:     sarama.StringEncoder(payload),
        Partition: int32(*partition),
    }
    fmt.Println("XXXX: ", msg.Partition)
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        panic(err)
    }
    fmt.Println()
    fmt.Printf("Message is stored in topic(%s)/partition(%d)/offset(%d)\n", *topic, partition, offset)
    fmt.Println("--------------------------------------------------")
    fmt.Println(partition)

}

问题描述

我正在尝试向我的 Kafka 代理发送消息流，我有 3 个指定的节点，我希望消息位于 12 个分区中，但是当我调用 SendMessage(msg) 时，它返回 0 个分区。这是为什么？为什么我不能在分区中发送消息？

4

1 回答 1

4

您尚未ManualPartitioner在配置中指定 ，因此它将使用默认值，而不是遵守您在分区字段中告诉它的内容。

添加到您的配置中

config.Producer.Partitioner = NewManualPartitioner

现在应该遵守您的分区值

（文档：https ://godoc.org/github.com/Shopify/sarama#example-Partitioner--Manual ）

于 2018-05-09T11:34:23.230 回答