kafka相关操作

- 启动
bin/zookeeper-server-start.sh config/zookeeper.properties

bin/kafka-server-start.sh config/server.properties

- 列出topicc
./kafka-topics.sh  --zookeeper 9.43.186.132:2181,9.43.186.152:2181,9.43.186.176:2181 --list 

- 创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic test --partitions 1 --replication-factor 1

- 查看topic的状态
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

- 消费者 读数据
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --group superpig

- 生产者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

- 删除topic
./kafka-topics --delete --zookeeper 10.0.8.23:2181 --topic PedesJobResult

/*
"github.com/Shopify/sarama"
cluster "github.com/bsm/sarama-cluster"
*/
ctx, cancel = context.WithCancel(context.Background())

config := cluster.NewConfig()
config.Consumer.Return.Errors = true
config.Group.Return.Notifications = true
config.Version = sarama.V2_0_0_0

// 这里很有迷惑性，实际上，这个选项只有第一次new consumer的时候才会有效，当partion已经存在offset，这是没用的
// 如果想每次重启，都忽略中间产生的消息，必须更换group_ip
config.Consumer.Offsets.Initial = sarama.OffsetNewest

var topicArr = []string{topic}
// KafkaAddresses=["kafka.service.consul:9092"]
consumer, err := cluster.NewConsumer(kafkaAddress,
    kafkaGroupID, topicArr, config)

if err != nil {
    logging.Errorf("cluster.NewConsumer  err:%s", err)
    return nil
}

go func() {
    for err := range consumer.Errors() {
        logging.Errorf("consumer.Error: groupId:%s:Error: %s\n", kafkaGroupID, err.Error())
    }
}()
go func() {
    for ntf := range consumer.Notifications() {
        logging.Infof("consumer.Notification: groupId:%s Rebalanced: %+v \n", kafkaGroupID, ntf)
    }
}()

logging.Infof("NewKafka loop before")
Loop:
    for {
        select {
        case msgc, ok := <-consumer.Messages():
            if ok {
                //logging.Debugf("read msg %v", msgc)
                // do sth
                // 如果sarama.OffsetNewest ，commit意义不大
                consumer.MarkOffset(msg, "")

            } else {
                logging.Errorf("read msg not ok %v", topic)
            }
        case <-ctx.Done():
            logging.Infof("kafka done %v", topic)
            break Loop
        case <-time.After(time.Second * 3):
            //logging.Debugf("NewKafka %v timeout", topic)
        }
    }
    logging.Infof("NewKafka kafka exit %v", topic)
    consumer.Close()

// producer config
config := sarama.NewConfig()
config.Producer.Retry.Max = 5
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Return.Successes = true

// sync producer
producer, err := sarama.NewSyncProducer(addr, config)

i := 0
for {
    i++
    msg := &sarama.ProducerMessage{
        Topic: topics[0],
        Value: sarama.StringEncoder(strconv.Itoa(i)),
    }
    _, _, err = producer.SendMessage(msg)
    checkerr.CheckError(err)
    time.Sleep(time.Millisecond * 500)
}