go - 在为 nats-streaming 设置处理程序时进行比赛数据检测

Question

对于 nats-streaming 订阅，处理程序需要如下所示

handler := func(ms *stan.Msg){}

我创建了以下函数，试图获取他们的处理程序，然后调用我自己的处理返回的消息。

func (t *T) Subscribe(sub Subscription, h ...message.Handler) (stan.Subscription, error) {
    process := func(ms *stan.Msg) {         
        for i := range h {
            h[i](ms)
        }
    }

    return t.subscribe(sub, process)
}

使用 go test -race 我在循环/调用我的处理程序的区域周围得到一个数据警告竞争条件。我正在寻找一些关于如何改进它以消除比赛警告的建议。

== 比赛检测输出

==================
WARNING: DATA RACE
Read at 0x00c4201f8040 by goroutine 82:
  github.com/MyTest/nats/v3_test.TestDurability.func2()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:399 +0x4d
  github.com/MyTest/nats/v3.(*T).Subscribe.func1()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:222 +0x2a0
  github.com/nats-io/go-nats-streaming.(*conn).processMsg()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/stan.go:464 +0x41c
  github.com/nats-io/go-nats-streaming.(*conn).(github.com/nats-io/go-nats-streaming.processMsg)-fm()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x4b
  github.com/nats-io/go-nats.(*Conn).waitForMsgs()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:1564 +0x1eb

Previous write at 0x00c4201f8040 by goroutine 81:
  github.com/MyTest/nats/v3_test.TestDurability.func1()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:383 +0xe7
  github.com/MyTest/nats/v3.(*T).Subscribe.func1()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:222 +0x2a0
  github.com/nats-io/go-nats-streaming.(*conn).processMsg()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/stan.go:464 +0x41c
  github.com/nats-io/go-nats-streaming.(*conn).(github.com/nats-io/go-nats-streaming.processMsg)-fm()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x4b
  github.com/nats-io/go-nats.(*Conn).waitForMsgs()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:1564 +0x1eb

Goroutine 82 (running) created at:
  github.com/nats-io/go-nats.(*Conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2238 +0x3f8
  github.com/nats-io/go-nats.(*Conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2155 +0x7e
  github.com/nats-io/go-nats-streaming.(*conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x59e
  github.com/nats-io/go-nats-streaming.(*conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:185 +0xb0
  github.com/MyTest/nats/v3.(*T).subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:279 +0x1be
  github.com/MyTest/nats/v3.(*T).Subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:226 +0x1f7
  github.com/MyTest/nats/v3_test.TestDurability()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:398 +0xa1a
  testing.tRunner()
      /usr/lib64/golang/src/testing/testing.go:746 +0x16c

Goroutine 81 (finished) created at:
  github.com/nats-io/go-nats.(*Conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2238 +0x3f8
  github.com/nats-io/go-nats.(*Conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2155 +0x7e
  github.com/nats-io/go-nats-streaming.(*conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x59e
  github.com/nats-io/go-nats-streaming.(*conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:185 +0xb0
  github.com/MyTest/nats/v3.(*T).subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:279 +0x1be
  github.com/MyTest/nats/v3.(*T).Subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:226 +0x1f7
  github.com/MyTest/nats/v3_test.TestDurability()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:382 +0x606
  testing.tRunner()
      /usr/lib64/golang/src/testing/testing.go:746 +0x16c
==================
==================
WARNING: DATA RACE
Read at 0x00c4201f0000 by goroutine 82:
  runtime.growslice()
      /usr/lib64/golang/src/runtime/slice.go:82 +0x0
  github.com/MyTest/nats/v3_test.TestDurability.func2()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:399 +0x166
  github.com/MyTest/nats/v3.(*T).Subscribe.func1()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:222 +0x2a0
  github.com/nats-io/go-nats-streaming.(*conn).processMsg()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/stan.go:464 +0x41c
  github.com/nats-io/go-nats-streaming.(*conn).(github.com/nats-io/go-nats-streaming.processMsg)-fm()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x4b
  github.com/nats-io/go-nats.(*Conn).waitForMsgs()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:1564 +0x1eb

Previous write at 0x00c4201f0000 by goroutine 81:
  github.com/MyTest/nats/v3_test.TestDurability.func1()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:383 +0xb2
  github.com/MyTest/nats/v3.(*T).Subscribe.func1()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:222 +0x2a0
  github.com/nats-io/go-nats-streaming.(*conn).processMsg()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/stan.go:464 +0x41c
  github.com/nats-io/go-nats-streaming.(*conn).(github.com/nats-io/go-nats-streaming.processMsg)-fm()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x4b
  github.com/nats-io/go-nats.(*Conn).waitForMsgs()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:1564 +0x1eb

Goroutine 82 (running) created at:
  github.com/nats-io/go-nats.(*Conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2238 +0x3f8
  github.com/nats-io/go-nats.(*Conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2155 +0x7e
  github.com/nats-io/go-nats-streaming.(*conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x59e
  github.com/nats-io/go-nats-streaming.(*conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:185 +0xb0
  github.com/MyTest/nats/v3.(*T).subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:279 +0x1be
  github.com/MyTest/nats/v3.(*T).Subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:226 +0x1f7
  github.com/MyTest/nats/v3_test.TestDurability()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:398 +0xa1a
  testing.tRunner()
      /usr/lib64/golang/src/testing/testing.go:746 +0x16c

Goroutine 81 (finished) created at:
  github.com/nats-io/go-nats.(*Conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2238 +0x3f8
  github.com/nats-io/go-nats.(*Conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats/nats.go:2155 +0x7e
  github.com/nats-io/go-nats-streaming.(*conn).subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:217 +0x59e
  github.com/nats-io/go-nats-streaming.(*conn).Subscribe()
      /home/chadit/Projects/src/github.com/nats-io/go-nats-streaming/sub.go:185 +0xb0
  github.com/MyTest/nats/v3.(*T).subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:279 +0x1be
  github.com/MyTest/nats/v3.(*T).Subscribe()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats.go:226 +0x1f7
  github.com/MyTest/nats/v3_test.TestDurability()
      /home/chadit/Projects/src/github.com/MyTest/nats/v3/nats_test.go:382 +0x606
  testing.tRunner()
      /usr/lib64/golang/src/testing/testing.go:746 +0x16c
==================
--- FAIL: TestDurability (1.01s)
    testing.go:699: race detected during execution of test
FAIL
exit status 1

链接到项目https://github.com/chadit/MessageQueueExamples/tree/master/nats-streaming/nats

go - 在为 nats-streaming 设置处理程序时进行比赛数据检测

0 回答 0

Related

Reference