关于go:用-Go-写一个简单消息队列四topic-设计

上一篇文章中咱们曾经实现了 channel 的设计，这里咱们持续实现 topic 的设计工作。

topic 的作用是接管客户端的音讯，而后同时发送给所有绑定的 channel 上，所以它的设计和 channel 很相似，蕴含的字段有：

name：名称
newChannelChan：新增 channel 的管道
channelMap：保护的 channel 汇合
incomingMessageChan：接管音讯的管道
msgChan：有缓冲管道，相当于音讯的内存队列
readSyncChan：和 routerSyncChan 配合应用保障 channelMap 的并发平安
routerSyncChan：见上
exitChan：接管退出信号的管道
channelWriteStarted：是否已向 channel 发送音讯

咱们须要保护一个全局的 topic map，在消费者订阅时生成新的 topic，相似于一个工厂，逻辑与第一篇中生成 uuid 相似：

注：Router 是 topic 的事件处理办法，详情见后文。

var (TopicMap     = make(map[string]*Topic)
    newTopicChan = make(chan util.ChanReq)
)

func NewTopic(name string, inMemSize int) *Topic {
    topic := &Topic{
        name:                name,
        newChannelChan:      make(chan util.ChanReq),
        channelMap:          make(map[string]*Channel),
        incomingMessageChan: make(chan *Message),
        msgChan:             make(chan *Message, inMemSize),
        readSyncChan:        make(chan struct{}),
        routerSyncChan:      make(chan struct{}),
        exitChan:            make(chan util.ChanReq),
    }
    go topic.Router(inMemSize)
    return topic
}

func GetTopic(name string) *Topic {topicChan := make(chan interface{})
    newTopicChan <- util.ChanReq{
        Variable: name,
        RetChan:  topicChan,
    }
    return (<-topicChan).(*Topic)
}

func TopicFactory(inMemSize int) {
    var (
        topicReq util.ChanReq
        name     string
        topic    *Topic
        ok       bool
    )
    for {
        topicReq = <-newTopicChan
        name = topicReq.Variable.(string)
        if topic, ok = TopicMap[name]; !ok {topic = NewTopic(name, inMemSize)
            TopicMap[name] = topic
            log.Printf("TOPIC %s CREATED", name)
        }
        topicReq.RetChan <- topic
    }
}

topic 保护 channel 的逻辑和 channel 保护消费者类似，也是“老熟人”chan + slice 的组合：

func (t *Topic) GetChannel(channelName string) *Channel {channelRet := make(chan interface{})
    t.newChannelChan <- util.ChanReq{
        Variable: channelName,
        RetChan:  channelRet,
    }
    return (<-channelRet).(*Channel)
}

func (t *Topic) Router(inMemSize int) {
    for {
        select {
        case channelReq := <-t.newChannelChan:
            channelName := channelReq.Variable.(string)
            channel, ok := t.channelMap[channelName]
            if !ok {channel = NewChannel(channelName, inMemSize)
                t.channelMap[channelName] = channel
                log.Printf("TOPIC(%s): new channel(%s)", t.name, channel.name)
            }
            channelReq.RetChan <- channel
        }
    }
}

此处的逻辑仍然与 channel 中的设计相似，间接贴代码：

func (t *Topic) PutMessage(msg *Message) {t.incomingMessageChan <- msg}

func (t *Topic) MessagePump() {
    var msg *Message
    for {
        select {case msg = <-t.msgChan:}

        t.readSyncChan <- struct{}{}

        for _, channel := range t.channelMap {go func(ch *Channel) {ch.PutMessage(msg)
            }(channel)
        }

        t.routerSyncChan <- struct{}{}
    }
}

func (t *Topic) Router(inMemSize int) {
    var (msg *Message)
    for {
        select {
        case channelReq := <-t.newChannelChan:
            ...
            if !t.channelWriteStarted {go t.MessagePump(closeChan)
                t.channelWriteStarted = true
            }
        case msg = <-t.incomingMessageChan:
            select {
            case t.msgChan <- msg:
                log.Printf("TOPIC(%s) wrote message", t.name)
            default:
            }
        case <-t.readSyncChan:
            <-t.routerSyncChan
        }
    }
}

咱们还是从 incomingMessageChan 中读取音讯，而后写入 msgChan，msgChan 缓冲区满了就抛弃（后续会加上长久化磁盘性能）。推送音讯到 channel 的协程是在增加 channel 时开启的，因为没有 channel 的话 topic 并不会推送音讯。

在向所有 channel 推送音讯的前后，咱们发现多了两个读管道的操作，这样做的目标是防止 map 的并发读写谬误。在 Go 语言中 map 是不反对并发读写的，因而咱们在遍历 channel 之前先读取 readSyncChan，确保咱们在遍历的时候调度协程是阻塞在 <-t.readSyncChan 这个 case 上，防止了对 map 的并发写操作。

敞开操作置信大家曾经一目了然了，无非就是监听推出信号的管道而后敞开 channel 和推送音讯的协程，代码如下：

func (t *Topic) MessagePump(closeChan <-chan struct{}) {
    ...
    for {
        select {
        ...
        case <-closeChan:
            return
        }
        ...
    }
}

func (t *Topic) Router(inMemSize int) {
    var (
        msg       *Message
        closeChan = make(chan struct{})
    )
    for {
        select {
        case channelReq := <-t.newChannelChan:
            ...
            if !t.channelWriteStarted {go t.MessagePump(closeChan)
                t.channelWriteStarted = true
            }
            ...
        case closeReq := <-t.exitChan:
            log.Printf("TOPIC(%s): closing", t.name)

            for _, channel := range t.channelMap {err := channel.Close()
                if err != nil {log.Printf("ERROR: channel(%s) close - %s", channel.name, err.Error())
                }
            }

            close(closeChan)
            closeReq.RetChan <- nil
        }
    }
}

func (t *Topic) Close() error {errChan := make(chan interface{})
    t.exitChan <- util.ChanReq{RetChan: errChan,}
    err, _ := (<-errChan).(error)
    return err
}

topic 残缺代码：topic.go

到这里咱们的两个外围组件 topic 和 channel 就全副设计实现了，下一篇文章咱们持续实现协定和后盾队列的性能。

当前目录构造为：

关于go:用-Go-写一个简单消息队列四topic-设计

topic

字段设计

topic 工厂

保护 channel

推送音讯给 channel

敞开