kafka作为现在比较火的消息队列服务，其优势除了其性能上，最主要的是其订阅模式，可以多个消费者消费同一个消息
consumer.go
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
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
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 |
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) } |
程序本天成，妙手偶得之！我们只是代码的搬运工！
转载请注明：http://www.521php.com/archives/1972/
