fix(kafka): 调整客户端一批只会发送一条消息 (#17)

juniaoshaonian · web-flow · commit 51167081e41e · 2025-11-10T10:02:51.000+08:00
diff --git a/.github/workflows/go-fmt.yml b/.github/workflows/go-fmt.yml
@@ -34,7 +34,7 @@ jobs:
         run: |
           go install mvdan.cc/gofumpt@latest && \
           go install golang.org/x/tools/cmd/goimports@latest && \
-          go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.54.2
+          go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.62.0
 
       - name: Check
         run: |
diff --git a/go.mod b/go.mod
@@ -1,6 +1,6 @@
 module github.com/ecodeclub/mq-api
 
-go 1.21
+go 1.24.2
 
 require (
 	github.com/ecodeclub/ekit v0.0.8
diff --git a/internal/pkg/validator/validator_test.go b/internal/pkg/validator/validator_test.go
@@ -35,7 +35,6 @@ func TestIsValidTopic(t *testing.T) {
 		}
 
 		for _, topic := range validTopics {
-			topic := topic
 			t.Run(topic, func(t *testing.T) {
 				t.Parallel()
 				assert.True(t, validator.IsValidTopic(topic), "Expected topic to be valid: %s", topic)
@@ -61,7 +60,6 @@ func TestIsValidTopic(t *testing.T) {
 		}
 
 		for _, topic := range invalidTopics {
-			topic := topic
 			t.Run(topic, func(t *testing.T) {
 				t.Parallel()
 				assert.False(t, validator.IsValidTopic(topic), "Expected topic to be invalid: %s", topic)
diff --git a/kafka/balancer_test.go b/kafka/balancer_test.go
@@ -80,7 +80,6 @@ func TestSpecifiedPartitionBalancer(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
diff --git a/kafka/producer.go b/kafka/producer.go
@@ -30,6 +30,10 @@ import (
 	kafkago "github.com/segmentio/kafka-go"
 )
 
+const (
+	defaultBatchSize = 1
+)
+
 type Producer struct {
 	topic  string
 	writer *kafkago.Writer
@@ -45,9 +49,10 @@ func NewProducer(address []string, topic string, balancer kafkago.Balancer) *Pro
 		topic:  topic,
 		locker: &sync.RWMutex{},
 		writer: &kafkago.Writer{
-			Addr:     kafkago.TCP(address...),
-			Topic:    topic,
-			Balancer: balancer,
+			Addr:      kafkago.TCP(address...),
+			Topic:     topic,
+			Balancer:  balancer,
+			BatchSize: defaultBatchSize,
 		},
 		closed:    false,
 		closeOnce: &sync.Once{},
diff --git a/memory/consumer.go b/memory/consumer.go
@@ -16,7 +16,6 @@ package memory
 
 import (
 	"context"
-	"log"
 	"sync"
 	"time"
 
@@ -67,9 +66,7 @@ func (c *Consumer) eventLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			log.Printf("消费者 %s 开始消费数据", c.name)
 			c.consumeAndReport()
-			log.Printf("消费者 %s 结束消费数据", c.name)
 		case event, ok := <-c.receiveCh:
 			if !ok {
 				return
@@ -84,7 +81,6 @@ func (c *Consumer) consumeAndReport() {
 	for idx, record := range c.partitionRecords {
 		msgs := c.partitions[record.Index].getBatch(record.Offset, limit)
 		for _, msg := range msgs {
-			log.Printf("消费者 %s 消费数据 %v", c.name, msg)
 			c.msgCh <- msg
 		}
 		record.Offset += len(msgs)
@@ -98,7 +94,6 @@ func (c *Consumer) consumeAndReport() {
 		}
 		err := <-errCh
 		if err != nil {
-			log.Printf("上报偏移量失败：%v", err)
 			return
 		}
 		close(errCh)
@@ -111,14 +106,12 @@ func (c *Consumer) handle(event *Event) {
 	// 服务端发起的重新加入事件
 	case RejoinEvent:
 		// 消费者上报消费进度
-		log.Printf("消费者 %s开始上报消费进度", c.name)
 		c.reportCh <- &Event{
 			Type: RejoinAckEvent,
 			Data: c.partitionRecords,
 		}
 		// 设置消费进度
 		partitionInfo := <-c.receiveCh
-		log.Printf("消费者 %s接收到分区信息 %v", c.name, partitionInfo)
 		c.partitionRecords, _ = partitionInfo.Data.([]PartitionRecord)
 		// 返回设置完成的信号
 		c.reportCh <- &Event{
@@ -155,13 +148,11 @@ func (c *Consumer) Close() error {
 			Type: ExitGroupEvent,
 			Data: c.closeCh,
 		}
-		log.Printf("消费者 %s 准备关闭", c.name)
 		// 等待服务端退出完成
 		<-c.closeCh
 		// 关闭资源
 		close(c.receiveCh)
 		close(c.msgCh)
-		log.Printf("消费者 %s 关闭成功", c.name)
 	})
 
 	return nil
diff --git a/memory/consumergroup.go b/memory/consumergroup.go
@@ -16,7 +16,6 @@ package memory
 
 import (
 	"fmt"
-	"log"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -105,16 +104,13 @@ func (c *ConsumerGroup) eventHandler(name string, event *Event) {
 		var err error
 		err = c.reportOffset(data.Records)
 		data.ErrChan <- err
-		log.Printf("消费者%s上报offset成功", name)
 	case RejoinAckEvent:
 		// consumer响应重平衡信号返回的数据，返回的是当前所有分区的偏移量
 		records, _ := event.Data.([]PartitionRecord)
 		// 不管上报成不成功
 		_ = c.reportOffset(records)
-		log.Printf("消费者%s成功接受到重平衡信号，并上报offset", name)
 		c.balanceCh <- struct{}{}
 	case PartitionNotifyAckEvent:
-		log.Printf("消费者%s 成功设置分区信息", name)
 		c.balanceCh <- struct{}{}
 	}
 }
@@ -128,12 +124,9 @@ func (c *ConsumerGroup) exitGroup(name string, closeCh chan struct{}) {
 			time.Sleep(defaultSleepTime)
 			continue
 		}
-		log.Printf("消费者 %s 准备退出消费组", name)
 		c.consumers.Delete(name)
 		c.reBalance()
-		log.Printf("给消费者 %s 发送退出确认信号", name)
 		close(closeCh)
-		log.Printf("消费者 %s 成功退出消费组", name)
 		if !atomic.CompareAndSwapInt32(&c.status, StatusBalancing, StatusStable) {
 			atomic.CompareAndSwapInt32(&c.status, StatusStopping, StatusStop)
 		}
@@ -167,7 +160,7 @@ func (c *ConsumerGroup) Close() {
 }
 
 func (c *ConsumerGroup) close() {
-	c.consumers.Range(func(key string, value *Consumer) bool {
+	c.consumers.Range(func(_ string, value *Consumer) bool {
 		ch := make(chan struct{})
 		value.receiveCh <- &Event{
 			Type: CloseEvent,
@@ -180,39 +173,30 @@ func (c *ConsumerGroup) close() {
 
 // reBalance 单独使用该方法是并发不安全的
 func (c *ConsumerGroup) reBalance() {
-	log.Println("开始重平衡")
 	// 通知每一个消费者进行偏移量的上报
 	length := 0
 	consumers := make([]string, 0, consumerCap)
-	log.Println("开始给每个消费者，重平衡信号")
 	c.consumers.Range(func(key string, value *Consumer) bool {
-		log.Printf("开始通知消费者%s", key)
 		value.receiveCh <- &Event{
 			Type: RejoinEvent,
 		}
 		consumers = append(consumers, key)
 		length++
-		log.Printf("通知消费者%s成功", key)
 		return true
 	})
 	number := 0
-	log.Println("xxxxxxxxxx长度", length)
 	// 等待所有消费者都接收到信号，并上报自己offset
 	for length > 0 {
 		<-c.balanceCh
 		number++
 		if number != length {
-			log.Println("xxxxxxxxxx number", number)
 			continue
 		}
 		// 接收到所有信号
-		log.Println("所有消费者已经接受到重平衡请求，并上报了消费进度")
 		consumerMap := c.consumerPartitionAssigner.AssignPartition(consumers, len(c.partitions))
 		// 通知所有消费者分配
-		log.Println("开始分配分区")
 		for consumerName, partitions := range consumerMap {
 			// 查找消费者所属的channel
-			log.Printf("消费者 %s 消费 %v 分区", consumerName, partitions)
 			consumer, ok := c.consumers.Load(consumerName)
 			if ok {
 				// 往每个消费者的receive_channel发送partition的信息
@@ -232,7 +216,6 @@ func (c *ConsumerGroup) reBalance() {
 
 			}
 		}
-		log.Println("重平衡结束")
 		return
 	}
 }
@@ -250,7 +233,7 @@ func (c *ConsumerGroup) JoinGroup() (*Consumer, error) {
 		}
 
 		var length int
-		c.consumers.Range(func(key string, value *Consumer) bool {
+		c.consumers.Range(func(_ string, _ *Consumer) bool {
 			length++
 			return true
 		})
@@ -269,7 +252,6 @@ func (c *ConsumerGroup) JoinGroup() (*Consumer, error) {
 		c.consumers.Store(name, consumer)
 		go c.consumerEventsHandler(name, reportCh)
 		go consumer.eventLoop()
-		log.Printf("新建消费者 %s", name)
 		// 重平衡分配分区
 		c.reBalance()
 		atomic.CompareAndSwapInt32(&c.status, StatusBalancing, StatusStable)
@@ -287,10 +269,3 @@ func (c *ConsumerGroup) consumerEventsHandler(name string, reportCh chan *Event)
 		}
 	}
 }
-
-func min(i, j int) int {
-	if i < j {
-		return i
-	}
-	return j
-}
diff --git a/memory/consumergroup_test.go b/memory/consumergroup_test.go
@@ -71,13 +71,13 @@ func TestConsumerGroup_Close(t *testing.T) {
 	wg.Wait()
 	// consumerGroup中没有消费者
 	var flag atomic.Bool
-	cg.consumers.Range(func(key string, value *Consumer) bool {
+	cg.consumers.Range(func(_ string, _ *Consumer) bool {
 		flag.Store(true)
 		return true
 	})
 	assert.False(t, flag.Load())
 	// 所有加入的消费者都是关闭状态
-	cg.consumers.Range(func(key string, value *Consumer) bool {
+	cg.consumers.Range(func(_ string, value *Consumer) bool {
 		assert.True(t, value.closed)
 		return true
 	})
diff --git a/memory/consumerpartitionassigner/equaldivide/balancer_test.go b/memory/consumerpartitionassigner/equaldivide/balancer_test.go
@@ -63,7 +63,6 @@ func TestBalancer_AssignPartition(t *testing.T) {
 		},
 	}
 	for _, tc := range testcases {
-		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 			actualVal := balancer.AssignPartition(tc.consumers, tc.partition)
diff --git a/memory/mq.go b/memory/mq.go
@@ -17,7 +17,7 @@ package memory
 import (
 	"context"
 	"fmt"
-	"log"
+	"log/slog"
 	"sync"
 
 	"github.com/ecodeclub/mq-api/internal/pkg/validator"
@@ -133,7 +133,7 @@ func (m *MQ) Close() error {
 	m.topics.Range(func(key string, value *Topic) bool {
 		err := value.Close()
 		if err != nil {
-			log.Printf("topic: %s关闭失败 %v", key, err)
+			slog.Error("topic关闭失败", slog.String("topic", key), slog.String("error", err.Error()))
 		}
 		return true
 	})
@@ -155,7 +155,7 @@ func (m *MQ) DeleteTopics(ctx context.Context, topics ...string) error {
 		if ok {
 			err := topic.Close()
 			if err != nil {
-				log.Printf("topic: %s关闭失败 %v", t, err)
+				slog.Error("topic关闭失败", slog.String("error", err.Error()))
 				continue
 			}
 			m.topics.Delete(t)
diff --git a/memory/produceridgetter/hash/get.go b/memory/produceridgetter/hash/get.go
@@ -29,5 +29,5 @@ func hashString(s string, numBuckets int) int64 {
 	h := fnv.New32a()
 	h.Write([]byte(s))
 	hash := h.Sum32()
-	return int64(hash % uint32(numBuckets))
+	return int64(int(hash) % numBuckets)
 }
diff --git a/memory/topic.go b/memory/topic.go
@@ -15,7 +15,6 @@
 package memory
 
 import (
-	"log"
 	"sync"
 
 	"github.com/ecodeclub/ekit/syncx"
@@ -76,7 +75,6 @@ func (t *Topic) addMessageWithPartition(msg *mq.Message, partitionID int64) erro
 	msg.Topic = t.name
 	msg.Partition = partitionID
 	t.partitions[partitionID].append(msg)
-	log.Printf("生产消息 %s,消息为 %s", t.name, msg.Value)
 	return nil
 }
 
@@ -85,7 +83,7 @@ func (t *Topic) Close() error {
 	defer t.locker.Unlock()
 	if !t.closed {
 		t.closed = true
-		t.consumerGroups.Range(func(key string, value *ConsumerGroup) bool {
+		t.consumerGroups.Range(func(_ string, value *ConsumerGroup) bool {
 			value.Close()
 			return true
 		})
diff --git a/scripts/deploy/dev-compose.yaml b/scripts/deploy/dev-compose.yaml
@@ -1,7 +1,7 @@
 version: "3"
 services:
   kafka0:
-   image: 'bitnami/kafka:3.6.0'
+   image: 'bitnamilegacy/kafka:3.6.0'
    ports:
      - '9092:9092'
      - '9094:9094'
diff --git a/scripts/lint/.golangci.yaml b/scripts/lint/.golangci.yaml
diff --git a/scripts/setup/setup.sh b/scripts/setup/setup.sh

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,6 @@ func TestIsValidTopic(t *testing.T) {`
`35`	`35`	`}`
`36`	`36`
`37`	`37`	`for _, topic := range validTopics {`
`38`		`- topic := topic`
`39`	`38`	`t.Run(topic, func(t *testing.T) {`
`40`	`39`	`t.Parallel()`
`41`	`40`	`assert.True(t, validator.IsValidTopic(topic), "Expected topic to be valid: %s", topic)`
`@@ -61,7 +60,6 @@ func TestIsValidTopic(t *testing.T) {`
`61`	`60`	`}`
`62`	`61`
`63`	`62`	`for _, topic := range invalidTopics {`
`64`		`- topic := topic`
`65`	`63`	`t.Run(topic, func(t *testing.T) {`
`66`	`64`	`t.Parallel()`
`67`	`65`	`assert.False(t, validator.IsValidTopic(topic), "Expected topic to be invalid: %s", topic)`
Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,6 @@ func TestSpecifiedPartitionBalancer(t *testing.T) {`
`80`	`80`	`}`
`81`	`81`
`82`	`82`	`for _, tc := range testCases {`
`83`		`- tc := tc`
`84`	`83`	`t.Run(tc.name, func(t *testing.T) {`
`85`	`84`	`t.Parallel()`
`86`	`85`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,6 @@ func TestBalancer_AssignPartition(t *testing.T) {`
`63`	`63`	`},`
`64`	`64`	`}`
`65`	`65`	`for _, tc := range testcases {`
`66`		`- tc := tc`
`67`	`66`	`t.Run(tc.name, func(t *testing.T) {`
`68`	`67`	`t.Parallel()`
`69`	`68`	`actualVal := balancer.AssignPartition(tc.consumers, tc.partition)`