Sarama: 如何进行分区，请大神指点下

[ruixule]

如何进行分区，请大神指点下

[ruixule]

1、之前无论采取哪种分区模式都无法进行
生产者的分区的分割器
分区选择在多个分区存在的情况下,决定将消息发送到哪个分区.
sarama有多个分割器:

sarama.NewManualPartitioner() //返回一个手动选择分区的分割器,也就是获取msg中指定的partition
sarama.NewRandomPartitioner() //通过随机函数随机获取一个分区号
sarama.NewRoundRobinPartitioner() //环形选择,也就是在所有分区中循环选择一个(徐工)
sarama.NewHashPartitioner() //通过msg中的key生成hash值,选择分区,

徐工 我建议是第3种，刚好均匀分区，目前我已经修改为5个默认分区，到时候可以均匀分配，

参考文献
https://blog.csdn.net/qq_32292967/article/details/78675116

2、servser.properties中需要将默认分区配置修改为你想设置的那个数。

3、测试代码
测试代码

/*
* @Author: Rui XuLe
* @Date:   2019-07-12 22:40:03
* @Last Modified by:   Rui XuLe
* @Last Modified time: 2019-07-13 22:05:08
 */
package main

import (
    "encoding/json"
    "fmt"
    "log"
    // "math/rand"
    "os"
    "time"

    "github.com/Shopify/sarama"
)

var Address = []string{"IP:9092"}

func main() {
    p := syncProducerInit(Address)
    sendMeg(p)
}

type MegInfo struct {
    Name    string    `json:"name"`
    Meg     string    `json:"meg"`
    TimeMeg time.Time `json:"timeTime"`
}

//同步消息模式
func syncProducerInit(address []string) (p sarama.SyncProducer) {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    config.Producer.Timeout = 5 * time.Second
    config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
    //sarama.NewHashPartitioner()
    p, err := sarama.NewSyncProducer(address, config)
    if err != nil {
        log.Printf("sarama.NewSyncProducer err, message=%s \n", err)
        return nil
    }
    return
}

//主要是 消息的发送
func sendMeg(p sarama.SyncProducer) {
    defer p.Close()
    topic := "test_go_kafka_producer"
    m := MegInfo{
        Name:    "RUIXULE",
        Meg:     "今天是20190720，我在测试gokafka",
        TimeMeg: time.Now(),
    }

    data, _ := json.Marshal(m)
    // srcValue := string(data) //"sync: this is ; message. index=%d"
    for i := 0; i < 100; i++ {
        value := string(data) // fmt.Sprintf(srcValue)
        msg := &sarama.ProducerMessage{
            //Partition: int32(i),
            //Key:       sarama.StringEncoder(fmt.Sprintf("%d", rand.Intn(10))),
            Timestamp: time.Now(),
            Topic:     topic,
            Value:     sarama.ByteEncoder(value),
        }
        part, offset, err := p.SendMessage(msg)
        if err != nil {
            log.Printf("send message(%s) err=%s \n", value, err)
        } else {
            fmt.Fprintf(os.Stdout, value+"发送成功，partition=%d, offset=%d \n", part, offset)
        }
    }
}

[FrancoisPoinsot]

By default, the hashPartitionner is used. It use the key of the message to define on which partition the message goes.

If you use roundRobin or random you lose one nice feature of kafka: guarantee of order.
When you use hashPartitionner, you have the guarantee that messages are produced and consumed in the correct order for any given key.
This key is whatever you want.
Example: if the key is the userID, for a given user, every event are ordered.

It might not matter, this totally depend on what you do.
To be honest hashPartionner is fast enough, I don't really see the point of using anything else.

Also a caveate: if you use hashPartitionner but set the key to nil (or just do not define it), the roundRobin partitioner will be used.

As for the partition count, 12 is a good number. It is sufficiently high for most use case and divide evenly in multiples ways ( 2 consumers, 6 partitions each; 3 consumers, 4 partitions each ...).

[d1egoaz]

I formatted the issue to help the reviewers

[d1egoaz]

@2017RXL even that nowadays is easy for use to use an online translator, could you please provide an English description to the problem, so reviewers won't need to use an online translator? this will also help you getting more 👀 as no many people will try to translate the message or don't speak Chinese as apparently @FrancoisPoinsot does 😄

[probot-shopify[bot]]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur.
Please check if the master branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.