consumer.go

  1. package main
  3. import (
  4.     "fmt"
  5.     "github.com/Shopify/sarama"
  6.     "log"
  7.     "os"
  8.     "strings"
  9.     "sync"
  10. )
  12. var (
  13.     wg     sync.WaitGroup
  14.     logger = log.New(os.Stderr, "[srama]", log.LstdFlags)
  15. )
  17. func main() {
  19.     sarama.Logger = logger
  21.     consumer, err := sarama.NewConsumer(strings.Split("localhost:9092", ","), nil)
  22.     if err != nil {
  23.         logger.Println("Failed to start consumer: %s", err)
  24.     }
  26.     partitionList, err := consumer.Partitions("hello")
  27.     if err != nil {
  28.         logger.Println("Failed to get the list of partitions: ", err)
  29.     }
  31.     for partition := range partitionList {
  32.         pc, err := consumer.ConsumePartition("hello", int32(partition), sarama.OffsetNewest)
  33.         if err != nil {
  34.             logger.Printf("Failed to start consumer for partition %d: %s\n", partition, err)
  35.         }
  36.         defer pc.AsyncClose()
  38.         wg.Add(1)
  40.         go func(sarama.PartitionConsumer) {
  41.             defer wg.Done()
  42.             for msg := range pc.Messages() {
  43.                 fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
  44.                 fmt.Println()
  45.             }
  46.         }(pc)
  47.     }
  49.     wg.Wait()
  51.     logger.Println("Done consuming topic hello")
  52.     consumer.Close()
  53. }

producer.go

  1. package main
  3. import (
  4.     "github.com/Shopify/sarama"
  5.     "log"
  6.     "os"
  7.     "strings"
  8. )
  10. var (
  11.     logger = log.New(os.Stderr, "[srama]", log.LstdFlags)
  12. )
  14. func main() {
  15.     sarama.Logger = logger
  17.     config := sarama.NewConfig()
  18.     config.Producer.RequiredAcks = sarama.WaitForAll
  19.     config.Producer.Partitioner = sarama.NewRandomPartitioner
  21.     msg := &sarama.ProducerMessage{}
  22.     msg.Topic = "hello"
  23.     msg.Partition = int32(-1)
  24.     msg.Key = sarama.StringEncoder("key")
  25.     msg.Value = sarama.ByteEncoder("你好, 世界!")
  27.     producer, err := sarama.NewSyncProducer(strings.Split("localhost:9092", ","), config)
  28.     if err != nil {
  29.         logger.Println("Failed to produce message: %s", err)
  30.         os.Exit(500)
  31.     }
  32.     defer producer.Close()
  34.     partition, offset, err := producer.SendMessage(msg)
  35.     if err != nil {
  36.         logger.Println("Failed to produce message: ", err)
  37.     }
  38.     logger.Printf("partition=%d, offset=%d\n", partition, offset)
  39. }