并发队列返回通道，锁定怀疑

Question

There is queue of not important structs Message, which has the classic push and pop methods:

type Queue struct {
    messages list.List
}

//The implementation is not relevant for the sake of the question
func (q *Queue) Push(msg Message) { /*...*/ }
func (q *Queue) Pop() (Message, bool) { /*...*/ }

/*
 * NewTimedChannel runs a goroutine which pops a message from the queue every 
 * given time duration and sends it over the returned channel 
 */
func (q *Queue) NewTimedChannel(t time.Duration) (<-chan Message) {/*...*/}

The client of the Push function will be a web gui in which users will post their messages.
The client of the channel returned by NewTimedChannel will be a service which sends each message to a not relevant endpoint over the network.

I'm a newbie in concurrency and go and I have the following question:

I know that since Queue.messages is a shared state between the main goroutine which deals with pushing the message after the user submit a web form and the ones created for each NewTimedChannel invocation, I need to lock it.

Do I need to lock and unlock using the sync.Mutex in all the Push, Pop and NewTimedChannel methods?
And is there a more idiomatic way to handle this specific problem in the go environment?

Answer 1

As others have pointed out, it requires synchronization or there will be a data race.

There is a saying in Go, "Don't communicate by sharing memory, share memory by communicating." As in this case, I think an idomatic way is to make channels send to a seprate goroutine which synchronize all the operations together using select. The code can easily be extended by adding more channels to support more kinds of operations (like the timed channel in your code which I don't fully understand what does it do), and by using select and other utils, it can easily handle more complex synchronizing by using locks. I write some sample code:

type SyncQueue struct {
    Q AbsQueue
    pushCh,popMsgCh chan Message
    popOkCh chan bool
    popCh chan struct{}
}

// An abstract of the Queue type. You can remove the abstract layer.
type AbsQueue interface {
    Push(Message)
    Pop() (Message,bool)
} 

func (sq SyncQueue) Push(m Message) {
    sq.pushCh <- m
}

func (sq SyncQueue) Pop() (Message,bool) {
    sq.popCh <- struct{}{} // send a signal for pop. struct{}{} cost no memory at all.

    return <-sq.popMsgCh,<-sq.popOkCh
}

// Every pop and push get synchronized here.
func (sq SyncQueue) Run() {
    for {
        select {
        case m:=<-pushCh:
            Q.Push(m)
        case <-popCh:
            m,ok := Q.Pop()
            sq.popMsgCh <- m
            sq.popOkCh <- ok
        }   
    }
}

func NewSyncQueue(Q AbsQueue) *SyncQueue {
    sq:=SyncQueue {
        Q:Q,
        pushCh: make(chan Message),popMsgCh: make(chan Message),
        pushOkCh: make(chan bool), popCh: make(chan struct{}),
    }
    go sq.Run()
    return &sq 
}

Note that for simpilicity, I did not use a quit channel or a context.Context, so the goroutine of sq.Run() has no way of exiting and would cause a memory leak.