设计模式：参与者风格

概述

参与者风格将问题分解为问题域相关的对象，每个对象中存在一个队列，并且暴露唯一的send接口用于给队列添加消息。对象轮循队列，并根据取出的不同消息执行不同的操作。

此风格适用于大型分布式系统中。每个组件不共享内存，使用消息进行功能的交互。

模式

下面实现一个词频统计器，即统计一段文章中top25的词频。主要结构如下： WordFrequencyController 管理者，它将文章切成词，然后逐个发送给StopWordManager

StopWordManager 包含一个停止词表。用来过滤停止词，如文章中的is, are等词为停止词，不进行词汇统计。如果接收到的词不在停止词中，则发送给词频统计器

WordFrequencyManager 包含词频表，接收到的词进行词频增加。并可以接收print命令。

大概的流程是：

WordFrequencyController  --send--> StopWordManager --send--> WordFrequencyManager

每个结构体对象都运行在不同的线程中，使用暴露的Send方法进行通讯。

具体代码

package main

import (
    "fmt"
    "strings"
    "time"
)

func main() {

    manager := new(WordFrequencyManager)
    manager.queue = make(chan []interface{})
    manager.wordMap = make(map[string]int)
    go manager.Run()

    stopWordManager := new(StopWordManager)
    stopWordManager.queue = make(chan []interface{})
    stopWordManager.manager = manager
    stopWordManager.stopWords = []string{"be", "is", "by", "and", "in", "of", "are", "to", "if", "on", "for", "the"}
    go stopWordManager.Run()

    controller := new(WordFrequencyController)
    controller.queue = make(chan []interface{})
    controller.manager = stopWordManager
    go controller.Run()

    controller.Send([]interface{}{"send_words", originData})
    time.Sleep(time.Millisecond * 300)
    manager.Send([]interface{}{"print"})
    time.Sleep(time.Millisecond * 300)
}

type WordFrequencyController struct {
    queue   chan []interface{}
    manager *StopWordManager
}

func (w *WordFrequencyController) Send(msg []interface{}) {
    w.queue <- msg
}

func (w *WordFrequencyController) Run() {
    for msg := range w.queue {
        ttype := msg[0]
        wordsStr := msg[1].(string)
        if ttype == "send_words" {
            words := w.split(wordsStr)
            for _, word := range words {
                w.manager.Send([]interface{}{"filter", word})
            }
        }
    }
}

func (w *WordFrequencyController) split(words string) []string {
    tpdata := strings.Split(words, " ")
    ret := make([]string, 0)
    for _, word := range tpdata {
        word = strings.Replace(word, ".", "", -1)
        word = strings.Replace(word, "'", "", -1)
        word = strings.Replace(word, ",", "", -1)
        word = strings.Replace(word, " ", "", -1)
        word = strings.ToLower(word)
        ret = append(ret, word)
    }
    return ret
}

type WordFrequencyManager struct {
    queue   chan []interface{}
    wordMap map[string]int
}

func (w *WordFrequencyManager) Send(msg []interface{}) {
    w.queue <- msg
}

func (w *WordFrequencyManager) Run() {
    for msg := range w.queue {
        ttype := msg[0]
        if ttype == "word" {
            word := msg[1].(string)
            v, ok := w.wordMap[word]
            if ok {
                w.wordMap[word] = v + 1
            } else {
                w.wordMap[word] = 1
            }
        } else if ttype == "print" {
            fmt.Println(w.wordMap)
        }
    }
}

type StopWordManager struct {
    queue     chan []interface{}
    stopWords []string
    manager   *WordFrequencyManager
}

func (s *StopWordManager) Send(msg []interface{}) {
    s.queue <- msg
}

func (s *StopWordManager) Run() {
    for msg := range s.queue {
        ttype := msg[0].(string)
        word := msg[1].(string)
        if ttype == "filter" {
            if !StrIn(word, s.stopWords) {
                s.manager.Send([]interface{}{"word", word})
            }
        }
    }
}

func StrIn(w string, l []string) bool {
    for _, item := range l {
        if w == item {
            return true
        }
    }
    return false
}

var originData = `The target field within a path is name for the target. This field MUST only ever be present on prefix paths in the corresponding request and response messages. This field is optional for clients. When set in the prefix in a request, GetRequest, SetRequest or SubscribeRequest, the field MUST be reflected in the prefix of the corresponding GetResponse, SetResponse or SubscribeResponse by a server. This field is used to allow a name to be associated with all the data for a given stream if requested by a client. If a client does not set this field in the prefix of a request, it MUST NOT be set in the prefix of the corresponding response messages. The value for target is tied to the context of a client RPC and not persisted or shared among multiple clients.`