107 lines
2.5 KiB
Go
107 lines
2.5 KiB
Go
package event_test
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"net"
|
||
"strconv"
|
||
"testing"
|
||
|
||
"github.com/segmentio/kafka-go"
|
||
)
|
||
|
||
// 创建Topic
|
||
func TestCreateTopic(t *testing.T) {
|
||
// 1. 连上kafka
|
||
conn, err := kafka.Dial("tcp", "localhost:9092")
|
||
if err != nil {
|
||
panic(err.Error())
|
||
}
|
||
defer conn.Close()
|
||
|
||
// contoller 管理, 获取zk地址后，管理集群状态
|
||
controller, err := conn.Controller()
|
||
if err != nil {
|
||
panic(err.Error())
|
||
}
|
||
var controllerConn *kafka.Conn
|
||
controllerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
|
||
if err != nil {
|
||
panic(err.Error())
|
||
}
|
||
defer controllerConn.Close()
|
||
|
||
// contoller 集群的维护
|
||
err = controllerConn.CreateTopics(kafka.TopicConfig{Topic: "maudit_new", NumPartitions: 6, ReplicationFactor: 1})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
|
||
// 查询Topic列表
|
||
func TestListTopic(t *testing.T) {
|
||
// 1. 连上kafka
|
||
conn, err := kafka.Dial("tcp", "localhost:9092")
|
||
if err != nil {
|
||
panic(err.Error())
|
||
}
|
||
defer conn.Close()
|
||
|
||
partitions, err := conn.ReadPartitions()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
topics := map[string]int{}
|
||
for _, p := range partitions {
|
||
topics[p.Topic]++
|
||
}
|
||
t.Log(topics)
|
||
}
|
||
|
||
func TestKafkaWirteMessage(t *testing.T) {
|
||
w := &kafka.Writer{
|
||
Addr: kafka.TCP("localhost:9092"),
|
||
Topic: "maudit_new",
|
||
Balancer: &kafka.LeastBytes{},
|
||
}
|
||
err := w.WriteMessages(context.Background(), kafka.Message{
|
||
// 支持 Writing to multiple topics
|
||
// NOTE: Each Message has Topic defined, otherwise an error is returned.
|
||
// Topic: "topic-A",
|
||
Key: []byte("Key-A"),
|
||
Value: []byte("Hello World!"),
|
||
},
|
||
kafka.Message{
|
||
Key: []byte("Key-B"),
|
||
Value: []byte("One!"),
|
||
},
|
||
kafka.Message{
|
||
Key: []byte("Key-C"),
|
||
Value: []byte("Two!"),
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
|
||
func TestKafkaReadMessage(t *testing.T) {
|
||
r := kafka.NewReader(kafka.ReaderConfig{
|
||
Brokers: []string{"localhost:9092"},
|
||
Topic: "maudit_new",
|
||
GroupID: "maudit",
|
||
})
|
||
// 自动化 1. 读取消息, 读出来 就代表已经被处理, FetchMessage, Commit(OK)
|
||
// r.ReadMessage(context.Background())
|
||
// 手动操作: 2. 获取消息, commit(OK), 对消息可靠度有要求，自己严格控制，避免消息丢失
|
||
for {
|
||
m, err := r.FetchMessage(context.Background())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
|
||
// 标记消息已处理
|
||
r.CommitMessages(context.Background(), m)
|
||
}
|
||
}
|