如何使用通道在go例程之间传递字节片

Question

我有一个函数，可以从源读取数据并将其发送到目标。源和目标可以是任何东西，假设这个示例的源是数据库（anyMySQL，PostgreSQL）目的地是分布式Q（任何…ActiveMQ，Kafka）。消息以字节存储。

这是主要功能。其想法是，它将旋转一个新的go例程，并等待消息返回以供将来处理。

type Message []byte

func (p *ProcessorService) Continue(dictId int) {
    level.Info(p.logger).Log("process", "message", "dictId", dictId)
    retrieved := make(chan Message)

    go func() {
        err := p.src.Read(retrieved, strconv.Itoa(p.dictId))
        if err != nil {
            level.Error(p.logger).Log("process", "read", "message", "err", err)
        }
    }()

    for r := range retrieved {
        go func(message Message) {
            level.Info(p.logger).Log("message", message)
            if len(message) > 0 {
                if err := p.dst.sendToQ(message); err != nil {
                    level.Error(p.logger).Log("failed", "during", "persist", "err", err)
                }
            } else {
                level.Error(p.logger).Log("failed")
            }
        }(r)
    }
}

这就是读取函数本身

func (s *Storage) Read(out chan<- Message, opt ...string) error {

    // I just skip some basic database read operations here
    // but idea is simple, read data from the table / file row by row and 
    // 
    for _, value := range dataFromDB {
            message, err := value.row 
            if err == nil {
                out <- message
            } else {
                errorf("Unable to get data %v", err)
                out <- make([]byte, 0)
            }
        }
    })

    close(out)

    if err != nil {
        return err
    }

    return nil
}

正如您所看到的，通过out chan完成通信

for r := range retrieved { 
   go func(message Message) {
       // basically here message and r are pointing to the same underlying array
   }
}

当接收到数据时，var r是一个类型的切片字节。然后它传递给go func（消息消息）go中按值传递的所有内容，在这种情况下var r将作为副本传递给匿名func，但是它仍然有一个指向底层片数据的指针。我很好奇它是否可能是一个问题在p.dst.sendToQ（消息）；执行，同时读取函数将发送一些东西到输出通道导致切片数据结构被一个新的信息覆盖。在传递给匿名函数之前，我应该将字节片r复制到新的字节片中吗？我测试了它，但不能真正导致这种行为。不确定我是偏执狂还是不得不担心。

Answer 1

当从数据库获取数据时，p.dst.sendToQ（消息）中的消息与value.row相同。所以，只要每个value.row有一个不同的底层数组，你就应该是好的。所以，我建议你检查源代码，确保它不使用公共字节数组，并不断重写到它。