对 broadcast.go 的 疑问

[nodephp]

broadcast.go

func (channel *Channel) Fetch(callback func(*Session)) {
    channel.mutex.RLock()
    defer channel.mutex.RUnlock()

    for _, sesssion := range channel.sessions {
        callback(sesssion.Session)
    }
}

callback 里面使用AsyncSend timeout 5s ，那么是不是这个广播也将 阻塞5s

[bg5sbk]

异步发送的代码是这样的：

func (session *Session) asyncSendBuffer(buffer *OutBuffer, timeout time.Duration) AsyncWork {
    c := make(chan error, 1)
    if session.IsClosed() {
        c <- SendToClosedError
    } else {
        select {
        case session.asyncSendBufferChan <- asyncBuffer{c, buffer}:
        default:
            if timeout == 0 {
                session.Close()
                c <- AsyncSendTimeoutError
            } else {
                go func() {
                    select {
                    case session.asyncSendBufferChan <- asyncBuffer{c, buffer}:
                    case <-session.closeChan:
                        c <- SendToClosedError
                    case <-time.After(timeout):
                        session.Close()
                        c <- AsyncSendTimeoutError
                    }
                }()
            }
        }
    }
    return AsyncWork{c}
}

这个设计的结果是：如果外部指定了timeout值为五秒，同时asyncSendBufferChan阻塞了，那么就会产生一个goroutine重新尝试发送，这个goroutine如果等待五秒还是发送不出去，就退出并返回错误。

因此在广播的时候，如果指定了timeout为五秒，也不会阻塞广播发送的过程，但是有可能短时间产生大量goroutine，所以建议广播的时候尽量不要给timeout值，如果要给也尽量不要太长，这样goroutine销毁比较快。

[bg5sbk]

广播最好别自己Fetch发送，已经有内置的广播接口。

// Broadcast to sessions. The message only encoded once
// so the performance is better than send message one by one.
func (b *Broadcaster) Broadcast(message Message, timeout time.Duration) ([]BroadcastWork, error) {
    buffer := newOutBuffer()
    b.protocol.PrepareOutBuffer(buffer, message.OutBufferSize())
    if err := message.WriteOutBuffer(buffer); err != nil {
        buffer.free()
        return nil, err
    }
    buffer.isBroadcast = true
    works := make([]BroadcastWork, 0, 10)
    b.fetcher(func(session *Session) {
        buffer.broadcastUse()
        works = append(works, BroadcastWork{
            session,
            session.asyncSendBuffer(buffer, timeout),
        })
    })
    return works, nil
}

[nodephp]

感谢达达 的解疑