golang，测试中的goroutines竞争条件

Question

我需要在发布到主题之前订阅一个主题（主题是一个频道），但是在创建一个线程时我需要运行 go Func 来继续监听频道以处理消息（例如从发布或订阅一个新的订阅 ） 测试有效（但不是每次都有效），有时当我运行测试时，它最终会在我收听主题（频道）之前在频道（主题）上发布一条消息

我有这个测试：

func Test_useCase_publish(t *testing.T) {
    for _, tt := range tests {
        tt := tt
        t.Run(tt.name, func(t *testing.T) {
            t.Parallel()

            tt.fields.storage = &RepositoryMock{
                GetTopicFunc: func(ctx context.Context, topicName vos.TopicName) (entities.Topic, error) {
                    return tt.fields.topic, nil
                },
            }

            useCase := New(tt.fields.storage)
            subscribed := make(chan struct{})
            go func() {
                tt.fields.topic.Activate()
                ch, _, err := useCase.Subscribe(tt.args.ctx, tt.args.message.TopicName)
                require.NoError(t, err)
                close(subscribed)
                msg, ok := <-ch
                if ok {
                    fmt.Println("msg", msg)
                    assert.Equal(t, tt.want, msg)
                }
            }()
            <-subscribed


            err := useCase.Publish(tt.args.ctx, tt.args.message)
            assert.ErrorIs(t, err, tt.wantErr)
        })
    }
}

主题：

func (t Topic) Activate() {
    go t.listenForSubscriptions()
    go t.listenForMessages()
    go t.listenForKills()
}

func (t *Topic) listenForSubscriptions() {
    for newSubCh := range t.newSubCh {
        t.Subscribers.Store(newSubCh.GetID(), newSubCh)
    }
}

func (t *Topic) listenForKills() {
    for subscriberID := range t.killSubCh {
        t.Subscribers.Delete(subscriberID)
    }
}

func (t *Topic) listenForMessages() {
    for msg := range t.newMessageCh {
        m := msg

        t.Subscribers.Range(func(key, value interface{}) bool {
            if key == nil || value == nil {
                return false
            }
            if subscriber, ok := value.(Subscriber); ok {
                subscriber.ReceiveMessage(m)
            }

            return true
        })
    }
func (t Topic) Dispatch(message vos.Message) {
    t.newMessageCh <- message
}
func (t *Topic) listenForMessages() {
    for msg := range t.newMessageCh {
        m := msg

        t.Subscribers.Range(func(key, value interface{}) bool {
            if key == nil || value == nil {
                return false
            }
            if subscriber, ok := value.(Subscriber); ok {
                subscriber.ReceiveMessage(m)
            }

            return true
        })
    }
}

订阅：

func (u useCase) Subscribe(ctx context.Context, topicName vos.TopicName) (chan vos.Message, vos.SubscriberID, error) {
    if err := topicName.Validate(); err != nil {
        return nil, "", err
    }

    topic, err := u.storage.GetTopic(ctx, topicName)
    if err != nil {
        if !errors.Is(err, entities.ErrTopicNotFound) {
            return nil, "", err
        }

        topic, err = u.createTopic(ctx, topicName)
        if err != nil {
            return nil, "", err
        }

        subscriber := entities.NewSubscriber(topic)

        subscriptionCh, id := subscriber.Subscribe()
        return subscriptionCh, id, nil
    }

    subscriber := entities.NewSubscriber(topic)

    subscriptionCh, id := subscriber.Subscribe()
    return subscriptionCh, id, nil
}

func (s Subscriber) Subscribe() (chan vos.Message, vos.SubscriberID) {
    s.topic.addSubscriber(s)
    return s.subscriptionCh, s.GetID()
}

func (s Subscriber) ReceiveMessage(msg vos.Message) {
    s.subscriptionCh <- msg
}

出版商：

func (u useCase) Publish(ctx context.Context, message vos.Message) error {
    if err := message.Validate(); err != nil {
        return err
    }

    topic, err := u.storage.GetTopic(ctx, message.TopicName)
    if err != nil {
        return err
    }

    topic.Dispatch(message)

    return nil
}

当我调用订阅时（我向订阅频道发送消息并添加订阅到我的线程）当我向主题发布消息时，我向主题频道发送消息

Answer 1

您显示的代码中缺少某些点，例如 .Subscribe() 和 .Publish() 的代码，或者通道是如何实例化的（它们是缓冲的还是非缓冲的？）。

---

可以说有一点：

从(t *Topic) listenForSubscriptions() 的外观来看：这种订阅方法不会向订阅者发送任何已注册的信号。

所以我的猜测是：您的useCase.Subscribe(...) 调用有创建的频道已写入newSubCH 的信息，但它没有得到t.Subcribers.Store(...) 已完成的信息。

因此，根据 goroutines 的调度方式，测试函数中的消息发送可能发生在通道实际注册之前。

要解决此问题，您可以添加一些将信号发送回调用方的内容。一种可能的方法：

type subscribeReq struct{
    ch   chan Message
    done chan struct{}
}

// turn Topic.newSubCh into a chan *subscribeReq
func (t *Topic) listenForSubscriptions() {
    for req := range t.newSubCh {
        t.Subscribers.Store(newSubCh.GetID(), req.ch)
        close(req.done)
    }
}

---

另外一点：你的测试函数根本不检查你的go func(){ ... }()调用的goroutine是否完成，所以你的单元测试过程也可能在goroutine有机会执行fmt.Println(msg)之前退出。

检查这一点的常用方法是使用sync.WaitGroup：

        t.Run(tt.name, func(t *testing.T) {
            ...
            useCase := New(tt.fields.storage)
            subscribed := make(chan struct{})
            wg := &sync.WaitGroup{}  // create a *sync.WaitGroup
            wg.Add(1)                // increment by 1 (you start only 1 goroutine)
            go func() {
               defer wg.Done()       // have the goroutine call wg.Done() when returning
               ...
            }()

            // send message, check that no error occurs
            wg.Wait()                // block here until the goroutine has completed
       })