协同程序和多线程有相似之处，但是又不是一个东西 协同程序是可以吧一个函数（routine） 分成几段执行，yield 关键字来等待协同程序的返回， 多线程在多核处理器的环境下能真正并行运行，协同程序任意时刻只有一个代码片段在运行 多线程要考虑线程同步，而协同程序不会，因为任意时刻只有一个协同程序的代码片段在运行 协同函数返回值必须是IEnumerator yield return 来返回 注意：4.3.1 最大允许协程数量为10000000个 1.用于定时器 启动1秒后debug输出 void Start () { StartCoroutine(func()); //立即返回 //StartCoroutine("func");//立即返回 } // Update is called once per frame void Update () { } IEnumerator func() { yield return new WaitForSeconds(1);//暂停当前函数片段，让出给其他片段执行，等待WaitForXXXXXX 返回 Debug.Log("1111111"); } 2.加载进度显示 public Text txt; // Use this for initialization bool is_done; int current; int total; void Start() { is_done = false; current = 0; total = 100; txt = GameObject.Find("Text").GetComponent(); StartCoroutine(loding_sync()); // StartCoroutine( ()=>IEnumerator{} ); } // Update is called once per frame void Update() { } IEnumerator loding_sync() { txt.text = "start to loaing"; yield return new WaitForSeconds(1.0f); for (; current <= total; ++current) { yield return new WaitForEndOfFrame(); txt.text = current.ToString() + "%"; dosomethingelse(); } txt.text = "Done"; } void dosomethingelse() { }

更像 订阅，项目需要，所以实现了一个简易的 using UnityEngine; using System.Collections; using System.Collections.Generic; public delegate void HandlerNotifyCenter(System.Object arg); [System.Serializable] class Recver { public string name; public GameObject recver; public HandlerNotifyCenter handler; public Recver() { } public void InVoke(System.Object arg) { handler(arg); } } public class NotifyCenter : MonoBehaviour { static public NotifyCenter ins; List recvers = new List(); void Awake() { ins = this; } // Use this for initialization void Start() { } // Update is called once per frame void Update() { } public void addRecver(GameObject recver, string name, HandlerNotifyCenter handler) { // gameObject.SendMessage(msg); Recver x = new Recver(); x.name = name; x.handler = handler; x.recver = recver; recvers.Add(x); } public void removeRecver(GameObject recver, string name) { foreach (Recver obj in recvers) { if (name == obj.name && recver == obj.recver) { recvers.Remove(obj); } } } public void sendMessage(string msg, System.Object arg) { foreach (Recver obj in recvers) { if (msg == obj.name) { obj.InVoke(arg); } } } }

计算top N words的topology, 用于比如trending topics or trending images on Twitter. 实现了滑动窗口计数和TopN排序, 比较有意思, 具体分析一下代码 Topology 这是一个稍微复杂些的topology, 主要体现在使用不同的grouping方式, fieldsGrouping和globalGrouping String spoutId = ""; String counterId = ""; String intermediateRankerId = ""; String totalRankerId = ""; builder.setSpout(spoutId, TestWordSpout(), 5); builder.setBolt(counterId, RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, Fields("")); builder.setBolt(intermediateRankerId, IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, Fields("")); builder.setBolt(totalRankerId, TotalRankingsBolt TOP_N)).globalGrouping(intermediateRankerId); RollingCountBolt 首先使用RollingCountBolt, 并且此处是按照word进行fieldsGrouping的, 所以相同的word会被发送到同一个bolt, 这个field id是在上一级的declareOutputFields时指定的 RollingCountBolt, 用于基于时间窗口的counting, 所以需要两个参数, the length of the sliding window in seconds和the emit frequency in seconds new RollingCountBolt(9, 3), 意味着output the latest 9 minutes sliding window every 3 minutes 1. 创建SlidingWindowCounter(SlidingWindowCounter和SlotBasedCounter参考下面) counter = new SlidingWindowCounter(this.windowLengthInSeconds / this.windowUpdateFrequencyInSeconds); 如何定义slot数? 对于9 min的时间窗口, 每3 min emit一次数据, 那么就需要9/3=3个slot 那么在3 min以内, 不停的调用countObjAndAck(tuple)来递增所有对象该slot上的计数 每3分钟会触发调用emitCurrentWindowCounts, 用于滑动窗口(通过getCountsThenAdvanceWindow), 并emit (Map, 实际使用时间) 因为实际emit触发时间, 不可能刚好是3 min, 会有误差, 所以需要给出实际使用时间 2. TupleHelpers.isTickTuple(tuple), TickTuple 前面没有说的一点是, 如何触发emit? 这是比较值得说明的一点, 因为其使用Storm的TickTuple特性. 这个功能挺有用, 比如数据库批量存储, 或者这里的时间窗口的统计等应用 "__system" component会定时往task发送 "__tick" stream的tuple 发送频率由TOPOLOGY_TICK_TUPLE_FREQ_SECS来配置, 可以在default.ymal里面配置 也可以在代码里面通过getComponentConfiguration()来进行配置, Map getComponentConfiguration() { Map conf = HashMap(); conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, emitFrequencyInSeconds); conf; 配置完成后, storm就会定期的往task发送ticktuple 只需要通过isTickTuple来判断是否为tickTuple, 就可以完成定时触发的功能 isTickTuple(Tuple tuple) { tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) \\ SYSTEM_COMPONENT_ID == "" && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID); \\ SYSTEM_TICK_STREAM_ID == "" } 最终, 这个blot的输出为, collector.emit(new Values(obj, count, actualWindowLengthInSeconds)); obj, count(窗口内的计数和), 实际使用时间 SlotBasedCounter 基于slot的counter, 模板类, 可以指定被计数对象的类型T 这个类其实很简单, 实现计数对象和一组slot(用long数组实现)的map, 并可以对任意slot做increment或reset等操作 关键结构为Map objToCounts, 为每个obj都对应于一个大小为numSlots的long数组, 所以对每个obj可以计numSlots个数 incrementCount, 递增某个obj的某个slot, 如果是第一次需要创建counts数组 getCount, getCounts, 获取某obj的某slot值, 或某obj的所有slot值的和 wipeSlot, resetSlotCountToZero, reset所有对象的某solt为0, reset某obj的某slot为0 wipeZeros, 删除所有total count为0的obj, 以释放空间 SlotBasedCounter Serializable { serialVersionUID = 4858185737378394432L; Map objToCounts = HashMap(); numSlots; SlotBasedCounter( numSlots) { (numSlots <= 0) { IllegalArgumentException("" + numSlots + ""); } .numSlots = numSlots; } incrementCount(T obj, slot) { [] counts = objToCounts.get(obj); (counts == ) { counts = [.numSlots]; objToCounts.put(obj, counts); } counts[slot]++; } getCount(T obj, slot) { [] counts = objToCounts.get(obj); (counts == ) { 0; } { counts[slot]; } } Map getCounts() { Map result = HashMap(); (T obj : objToCounts.keySet()) { result.put(obj, computeTotalCount(obj)); } result; } computeTotalCount(T obj) { [] curr = objToCounts.get(obj); total = 0; ( l : curr) { total += l; } total; } wipeSlot( slot) { (T obj : objToCounts.keySet()) { resetSlotCountToZero(obj, slot); } } resetSlotCountToZero(T obj, slot) { [] counts = objToCounts.get(obj); counts[slot] = 0; } shouldBeRemovedFromCounter(T obj) { computeTotalCount(obj) == 0; } wipeZeros() { Set objToBeRemoved = HashSet(); (T obj : objToCounts.keySet()) { (shouldBeRemovedFromCounter(obj)) { objToBeRemoved.add(obj); } } (T obj : objToBeRemoved) { objToCounts.remove(obj); } } } SlidingWindowCounter SlidingWindowCounter只是对SlotBasedCounter做了进一步的封装, 通过headSlot和tailSlot提供sliding window的概念 incrementCount, 只能对headSlot进行increment, 其他slot作为窗口中的历史数据 核心的操作为, getCountsThenAdvanceWindow 1. 取出Map counts, 对象和窗口内所有slots求和值的map 2. 调用wipeZeros, 删除已经不被使用的obj, 释放空间 3. 最重要的一步, 清除tailSlot, 并advanceHead, 以实现滑动窗口 advanceHead的实现, 如何在数组实现循环的滑动窗口 SlidingWindowCounter Serializable { serialVersionUID = -2645063988768785810L; SlotBasedCounter objCounter; headSlot; tailSlot; windowLengthInSlots; SlidingWindowCounter( windowLengthInSlots) { (windowLengthInSlots < 2) { IllegalArgumentException("" + windowLengthInSlots + ""); } .windowLengthInSlots = windowLengthInSlots; .objCounter = SlotBasedCounter(.windowLengthInSlots); .headSlot = 0; .tailSlot = slotAfter(headSlot); } incrementCount(T obj) { objCounter.incrementCount(obj, headSlot); } Map getCountsThenAdvanceWindow() { Map counts = objCounter.getCounts(); objCounter.wipeZeros(); objCounter.wipeSlot(tailSlot); advanceHead(); counts; } advanceHead() { headSlot = tailSlot; tailSlot = slotAfter(tailSlot); } slotAfter( slot) { (slot + 1) % windowLengthInSlots; } } IntermediateRankingsBolt 这个bolt作用就是对于中间结果的排序, 为什么要增加这步, 应为数据量比较大, 如果直接全放到一个节点上排序, 会负载太重 所以先通过IntermediateRankingsBolt, 过滤掉一些 这里仍然使用, 对于obj进行fieldsGrouping, 保证对于同一个obj, 不同时间段emit的统计数据会被发送到同一个task IntermediateRankingsBolt继承自AbstractRankerBolt(参考下面) 并实现了updateRankingsWithTuple, updateRankingsWithTuple(Tuple tuple) { Rankable rankable = RankableObjectWithFields.from(tuple); .getRankings().updateWith(rankable); } 逻辑很简单, 将Tuple转化Rankable, 并更新Rankings列表 参考AbstractRankerBolt, 该bolt会定时将Ranking列表emit出去 Rankable Rankable除了继承Comparable接口, 还增加getObject()和getCount()接口 Rankable Comparable { Object getObject(); getCount(); } RankableObjectWithFields RankableObjectWithFields实现Rankable接口 1. 提供将Tuple转化为RankableObject Tuple由若干field组成, 第一个field作为obj, 第二个field作为count, 其余的都放到List