我正在kafka中使用kafka go库构建一个请求-响应设置，使用消息键作为相关id。我的设置在没有并发的情况下工作正常，但是当消息开始在单独的goroutine中发送时，reader部分跳过正确的键（因为其他例程可能已经读取了它）。
考虑到连接被不同的goroutine共享，如何从一个主题中只读取一个特定的键？
下面的客户端示例（为了简洁起见，删除了错误评估）：

package main

import (
    "bytes"
    "context"
    "fmt"
    "sync"
    "time"

    "github.com/google/uuid"
    kafka "github.com/segmentio/kafka-go"
)

var wg sync.WaitGroup

func requestMessage(connR *kafka.Conn, connW *kafka.Conn, body []byte, index int) {
    currentUUID := uuid.New()
    byteUUID := []byte(fmt.Sprintf("%s", currentUUID))
    connW.WriteMessages(kafka.Message{
        Key:   byteUUID,
        Value: body,
    })
    fmt.Println("Posted id " + string(byteUUID))
    for {
        m, _ := connR.ReadMessage(10e6)
        if bytes.Equal(m.Key, byteUUID) {
            break
        }
    }

    wg.Done()
    fmt.Println("Done " + string(byteUUID))

}

func main() {
    iterations := 100
    interval := 500 * time.Millisecond
    kafkaURL := "kafka:9092"
    topic := "benchmarktopic"
    partition := 0
    connW, _ := kafka.DialLeader(context.Background(), "tcp", kafkaURL, topic, partition)
    defer connW.Close()
    connR, _ := kafka.DialLeader(context.Background(), "tcp", kafkaURL, topic+"response", partition)
    defer connR.Close()
    for i := 0; i < iterations; i++ {
        <-time.After(interval)
        go requestMessage(connR, connW, []byte("body"), i)
        wg.Add(1)
    }
    wg.Wait()
}