kafka consumer 消费

• 使用 kafka-go 库
• 安装 kafka-go 库

go get github.com/segmentio/kafka-go
// 使用账号密码认证
go get github.com/segmentio/kafka-go/sasl/scram@v0.4.32
1
2
3

package sasl_test

import (
	"context"
	"testing"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
	"github.com/segmentio/kafka-go/sasl/plain"
	"github.com/segmentio/kafka-go/sasl/scram"
	ktesting "github.com/segmentio/kafka-go/testing"
)

const (
	saslTestConnect = "localhost:9093" // connect to sasl listener
	saslTestTopic   = "test-writer-0"  // this topic is guaranteed to exist.
)

func TestSASL(t *testing.T) {
	tests := []struct {
		valid    func() sasl.Mechanism
		invalid  func() sasl.Mechanism
		minKafka string
	}{
		{
			valid: func() sasl.Mechanism {
				return plain.Mechanism{
					Username: "adminplain",
					Password: "admin-secret",
				}
			},
			invalid: func() sasl.Mechanism {
				return plain.Mechanism{
					Username: "adminplain",
					Password: "badpassword",
				}
			},
		},
		{
			valid: func() sasl.Mechanism {
				mech, _ := scram.Mechanism(scram.SHA256, "adminscram", "admin-secret-256")
				return mech
			},
			invalid: func() sasl.Mechanism {
				mech, _ := scram.Mechanism(scram.SHA256, "adminscram", "badpassword")
				return mech
			},
			minKafka: "0.10.2.0",
		},
		{
			valid: func() sasl.Mechanism {
				mech, _ := scram.Mechanism(scram.SHA512, "adminscram", "admin-secret-512")
				return mech
			},
			invalid: func() sasl.Mechanism {
				mech, _ := scram.Mechanism(scram.SHA512, "adminscram", "badpassword")
				return mech
			},
			minKafka: "0.10.2.0",
		},
	}

	for _, tt := range tests {
		mech := tt.valid()
		if !ktesting.KafkaIsAtLeast(tt.minKafka) {
			t.Skip("requires min kafka version " + tt.minKafka)
		}

		t.Run(mech.Name()+" success", func(t *testing.T) {
			testConnect(t, tt.valid(), true)
		})
		t.Run(mech.Name()+" failure", func(t *testing.T) {
			testConnect(t, tt.invalid(), false)
		})
		t.Run(mech.Name()+" is reusable", func(t *testing.T) {
			mech := tt.valid()
			testConnect(t, mech, true)
			testConnect(t, mech, true)
			testConnect(t, mech, true)
		})

	}
}

func testConnect(t *testing.T, mechanism sasl.Mechanism, success bool) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	d := kafka.Dialer{
		SASLMechanism: mechanism,
	}
	_, err := d.DialLeader(ctx, "tcp", saslTestConnect, saslTestTopic, 0)
	if success && err != nil {
		t.Errorf("should have logged in correctly, got err: %v", err)
	} else if !success && err == nil {
		t.Errorf("should not have logged in correctly")
	}
}
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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99