golang kafka

zhangxinba

　　consumer.go
package main import (    "fmt"    "github.com/Shopify/sarama"    "log"    "os"    "strings"    "sync") var (    wg     sync.WaitGroup    logger = log.New(os.Stderr, "[srama]", log.LstdFlags)) func main() {     sarama.Logger = logger     consumer, err := sarama.NewConsumer(strings.Split("localhost:9092", ","), nil)    if err != nil {        logger.Println("Failed to start consumer: %s", err)    }     partitionList, err := consumer.Partitions("hello")    if err != nil {        logger.Println("Failed to get the list of partitions: ", err)    }     for partition := range partitionList {        pc, err := consumer.ConsumePartition("hello", int32(partition), sarama.OffsetNewest)        if err != nil {            logger.Printf("Failed to start consumer for partition %d: %s\n", partition, err)        }        defer pc.AsyncClose()         wg.Add(1)         go func(sarama.PartitionConsumer) {            defer wg.Done()            for msg := range pc.Messages() {                fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))                fmt.Println()            }        }(pc)    }     wg.Wait()     logger.Println("Done consuming topic hello")    consumer.Close()}　　producer.go
package main import (    "github.com/Shopify/sarama"    "log"    "os"    "strings") var (    logger = log.New(os.Stderr, "[srama]", log.LstdFlags)) func main() {    sarama.Logger = logger     config := sarama.NewConfig()    config.Producer.RequiredAcks = sarama.WaitForAll    config.Producer.Partitioner = sarama.NewRandomPartitioner     msg := &sarama.ProducerMessage{}    msg.Topic = "hello"    msg.Partition = int32(-1)    msg.Key = sarama.StringEncoder("key")    msg.Value = sarama.ByteEncoder("你好, 世界!")     producer, err := sarama.NewSyncProducer(strings.Split("localhost:9092", ","), config)    if err != nil {        logger.Println("Failed to produce message: %s", err)        os.Exit(500)    }    defer producer.Close()     partition, offset, err := producer.SendMessage(msg)    if err != nil {        logger.Println("Failed to produce message: ", err)    }    logger.Printf("partition=%d, offset=%d\n", partition, offset)}