1辛燥，如何保證消息100%投遞镜沽？
--生產(chǎn)者
--消費者
2，如何保證消息的順序性茂洒？
kafka同一個分區(qū)可以保證順序性，因此只需要將需要保證順序性的消息放到同一個分區(qū)即可（要么只用一個分區(qū)似将，要么用hash分區(qū)）获黔。
3，如何保證消息不重復在验？
消息是有可能重復的玷氏，但是我們要保證消費了重復的消息也不出問題，就是要保證消息的冪等性
4腋舌，如何單播盏触、多播？
通過消費者組來實現(xiàn)。一個主題可以設置0或者多個消費者組赞辩，在同一個消費者組內(nèi)雌芽，不同消費者消費的內(nèi)容是互斥的。注意辨嗽，同一個消費者組內(nèi)的消費者個數(shù)不能大于分區(qū)數(shù)量世落，否則多余的消費者不會分配到分區(qū)將處于阻塞狀態(tài)
消費者
KafkaConsumer.go
package cws
import (
"github.com/IBM/sarama"
"log"
"os"
"os/signal"
)
func KafkaConsumer() {
config := sarama.NewConfig()
config.Consumer.Return.Errors = true
client, err := sarama.NewClient([]string{"localhost:9192", "localhost:9292", "localhost:9392"}, config)
defer client.Close()
if err != nil {
panic(err)
}
consumer, err := sarama.NewConsumerFromClient(client)
defer consumer.Close()
if err != nil {
panic(err)
}
// get partitionId list
partitions, err := consumer.Partitions("my_topic")
if err != nil {
panic(err)
}
for _, partitionId := range partitions {
// create partitionConsumer for every partitionId
partitionConsumer, err := consumer.ConsumePartition("my_topic", partitionId, sarama.OffsetNewest)
if err != nil {
panic(err)
}
go func(pc *sarama.PartitionConsumer) {
defer (*pc).Close()
// block
for message := range (*pc).Messages() {
value := string(message.Value)
log.Printf("Partitionid: %d; offset:%d, value: %s\n", message.Partition, message.Offset, value)
}
}(&partitionConsumer)
}
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt)
select {
case <-signals:
}
}
消費組
KafkaConsumerGroup.go
package cws
import (
"context"
"fmt"
"github.com/IBM/sarama"
"os"
"os/signal"
"sync"
)
type consumerGroupHandler struct {
name string
}
func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("%s Message topic:%q partition:%d offset:%d value:%s\n", h.name, msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
// 手動確認消息
sess.MarkMessage(msg, "")
//手動的話需要提交下。
sess.Commit()
}
return nil
}
func handleErrors(group *sarama.ConsumerGroup, wg *sync.WaitGroup) {
wg.Done()
for err := range (*group).Errors() {
fmt.Println("ERROR", err)
}
}
func consume(group *sarama.ConsumerGroup, wg *sync.WaitGroup, name string) {
fmt.Println(name + "start")
wg.Done()
ctx := context.Background()
for {
topics := []string{"my_topic"}
handler := consumerGroupHandler{name: name}
err := (*group).Consume(ctx, topics, handler)
if err != nil {
panic(err)
}
}
}
func KafkaConsumerGroup() {
var wg sync.WaitGroup
config := sarama.NewConfig()
config.Consumer.Return.Errors = false
config.Version = sarama.V0_10_2_0
client, err := sarama.NewClient([]string{"localhost:9192", "localhost:9292", "localhost:9392"}, config)
defer client.Close()
if err != nil {
panic(err)
}
group1, err := sarama.NewConsumerGroupFromClient("c1", client)
if err != nil {
panic(err)
}
group2, err := sarama.NewConsumerGroupFromClient("c2", client)
if err != nil {
panic(err)
}
group3, err := sarama.NewConsumerGroupFromClient("c3", client)
if err != nil {
panic(err)
}
defer group1.Close()
defer group2.Close()
defer group3.Close()
wg.Add(3)
go consume(&group1, &wg, "c1")
go consume(&group2, &wg, "c2")
go consume(&group3, &wg, "c3")
wg.Wait()
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt)
select {
case <-signals:
}
}
生產(chǎn)者
KafkaProducer.go
package cws
import (
"github.com/IBM/sarama"
"log"
"os"
"os/signal"
"sync"
)
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
config.Producer.Partitioner = sarama.NewRandomPartitioner
client, err := sarama.NewClient([]string{"192.168.0.104:9192", "localhost:9292", "localhost:9392"}, config)
defer client.Close()
if err != nil {
panic(err)
}
producer, err := sarama.NewAsyncProducerFromClient(client)
if err != nil {
panic(err)
}
// Trap SIGINT to trigger a graceful shutdown.
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt)
var (
wg sync.WaitGroup
enqueued, successes, errors int
)
wg.Add(1)
// start a groutines to count successes num
go func() {
defer wg.Done()
for range producer.Successes() {
successes++
}
}()
wg.Add(1)
// start a groutines to count error num
go func() {
defer wg.Done()
for err := range producer.Errors() {
log.Println(err)
errors++
}
}()
ProducerLoop:
for {
message := &sarama.ProducerMessage{Topic: "my_topic", Value: sarama.StringEncoder("testing 123")}
select {
case producer.Input() <- message:
enqueued++
case <-signals:
producer.AsyncClose() // Trigger a shutdown of the producer.
break ProducerLoop
}
}
wg.Wait()
log.Printf("Successfully produced: %d; errors: %d\n", successes, errors)
}
