kafka
消息记录(record): 由一个key，一个value和一个时间戳构成，消息最终存储在主题下的分区中, 记录在生产者中称为生产者记录(ProducerRecord), 在消费者中称为消费者记录(ConsumerRecord)，Kafka集群保持所有的消息，直到它们过期， 无论消息是否被消费了，在一个可配置的时间段内，Kafka集群保留所有发布的消息，不管这些消息有没有被消费。比如，如果消息的保存策略被设置为2天，那么在一个消息被发布的两天时间内，它都是可以被消费的。之后它将被丢弃以释放空间。Kafka的性能是和数据量无关的常量级的，所以保留太多的数据并不是问题。
生产者(producer): 生产者用于发布(send)消息
消费者(consumer): 消费者用于订阅(subscribe)消息
消费者组(consumer group): 相同的group.id的消费者将视为同一个消费者组, 每个消费者都需要设置一个组id, 每条消息只能被 consumer group 中的一个 Consumer 消费,但可以被多个 consumer group 消费
主题(topic): 消息的一种逻辑分组，用于对消息分门别类，每一类消息称之为一个主题，相同主题的消息放在一个队列中
分区(partition): 消息的一种物理分组， 一个主题被拆成多个分区，每一个分区就是一个顺序的、不可变的消息队列，并且可以持续添加，分区中的每个消息都被分配了一个唯一的id，称之为偏移量(offset),在每个分区中偏移量都是唯一的。每个分区对应一个逻辑log，有多个segment组成。
偏移量(offset): 分区中的每个消息都一个一个唯一id，称之为偏移量，它代表已经消费的位置。可以自动或者手动提交偏移量(即自动或者手动控制一条消息是否已经被成功消费)
代理(broker): 一台kafka服务器称之为一个broker
副本(replica)：副本只是一个分区(partition)的备份。 副本从不读取或写入数据。 它们用于防止数据丢失。
领导者(leader)：Leader 是负责给定分区的所有读取和写入的节点。 每个分区都有一个服务器充当Leader， producer 和 consumer 只跟 leader 交互
追随者(follower)：跟随领导者指令的节点被称为Follower。 如果领导失败，一个追随者将自动成为新的领导者。 跟随者作为正常消费者，拉取消息并更新其自己的数据存储。replica 中的一个角色，从 leader 中复制数据。
zookeeper：Kafka代理是无状态的，所以他们使用ZooKeeper来维护它们的集群状态。ZooKeeper用于管理和协调Kafka代理
kafka功能
发布订阅：生产者(producer)生产消息(数据流), 将消息发送到到kafka指定的主题队列(topic)中，也可以发送到topic中的指定分区(partition)中，消费者(consumer)从kafka的指定队列中获取消息，然后来处理消息。
流处理(Stream Process): 将输入topic转换数据流到输出topic
连接器(Connector) : 将数据从应用程序(源系统)中导入到kafka，或者从kafka导出数据到应用程序(宿主系统sink system), 例如：将文件中的数据导入到kafka，从kafka中将数据导出到文件中
kafka中的消息模型
队列：同名的消费者组员瓜分消息
发布订阅：广播消息给多个消费者组(不同名)
生产者(producer)将消息记录(record)发送到kafka中的主题中(topic), 一个主题可以有多个分区(partition), 消息最终存储在分区中，消费者(consumer)最终从主题的分区中获取消息。
使用库sarama 文档： https://godoc.org/github.com/Shopify/sarama#example-Consumer
package main
import (
"fmt"
"github.com/Shopify/sarama"
"github.com/bsm/sarama-cluster"
"os"
"os/signal"
"syscall"
"time"
)
func newSub() *cluster.Consumer {
config := cluster.NewConfig()
config.Consumer.Return.Errors = true
config.Group.Return.Notifications = true
config.Consumer.Offsets.Initial = sarama.OffsetNewest
config.Consumer.Offsets.CommitInterval=config.Consumer.Offsets.AutoCommit.Interval
consumer, err := cluster.NewConsumer([]string{"127.0.0.1:9092"}, "test-push-group", []string{"test-push-topic"}, config)
if err != nil {
panic(err)
}
return consumer
}
// Consume messages, watch signals
func Consumer(c *cluster.Consumer,code string) {
for {
select {
case err := <-c.Errors():
fmt.Errorf("consumer error(%v)\n", err)
case n := <-c.Notifications():
fmt.Printf("consumer rebalanced(%v)\n", n)
case msg, ok := <-c.Messages():
if !ok {
return
}
fmt.Printf("consume%s: key %s value %s\n", code,msg.Key,msg.Value)
}
}
}
func newSyncPub() sarama.SyncProducer {
kc := sarama.NewConfig()
kc.Producer.RequiredAcks = sarama.WaitForAll // Wait for all in-sync replicas to ack the message
kc.Producer.Retry.Max = 10 // Retry up to 10 times to produce the message
kc.Producer.Return.Successes = true
pub, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, kc)
if err != nil {
panic(err)
}
return pub
}
func SyncProducer(c sarama.SyncProducer){
for{
m := &sarama.ProducerMessage{
Key: sarama.StringEncoder("SyncProducer"),
Topic: "test-push-topic",
Value: sarama.ByteEncoder("ssssssssssss"),
}
if _, _, err := c.SendMessage(m); err != nil {
fmt.Errorf("PushMsg.send error(%v)\n", err)
}
fmt.Printf("SyncProducer send\n")
time.Sleep(time.Second * 10)
}
}
func newPub() sarama.AsyncProducer {
kc := sarama.NewConfig()
kc.Producer.Return.Successes = true //必须有这个选项
kc.Producer.Timeout = 5 * time.Second
pub, err := sarama.NewAsyncProducer([]string{"127.0.0.1:9092"}, kc)
if err != nil {
panic(err)
}
return pub
}
func Producer(p sarama.AsyncProducer){
//必须有这个匿名函数内容
go func(p sarama.AsyncProducer) {
errors := p.Errors()
success := p.Successes()
for {
select {
case err := <-errors:
if err != nil {
fmt.Println(err)
}
case <-success:
}
}
}(p)
for{
msg := &sarama.ProducerMessage{
Key: sarama.StringEncoder("Producer"),
Topic: "test-push-topic",
Value: sarama.ByteEncoder("pppppppppppppppp"),
}
p.Input() <- msg
fmt.Printf("Producer send\n")
time.Sleep(time.Second * 10)
}
}
func main(){
sub:=newSub()
go Consumer(sub,"1")
go Consumer(sub,"2")
go Consumer(sub,"3")
//同步生产者
pubSync:=newSyncPub()
go SyncProducer(pubSync)
//异步生产者
pub:=newPub()
go Producer(pub)
c := make(chan os.Signal, 1)
signal.Notify(c, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
for {
s := <-c
fmt.Printf("job get a signal %s", s.String())
switch s {
case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:
sub.Close()
pubSync.Close()
pub.Close()
fmt.Printf("job exit")
return
case syscall.SIGHUP:
default:
return
}
}
}