Python3.4模拟实现生产者消费者模

lra***

Worker算法执行于图中每个顶点之上，每个顶点有一个相关的消息游标和一个global对象，里面含有步长信息和用户定义的Global数据。该算法定义如下：

[js] view plaincopy

    var connectedSetsAlgorithm = function (vertex, message, global) {  

为了检测所有的节点组，这里使用了一种非常直接的方法：

每个节点组有一个字母标识符，存有其顶点最后的_key属性信息。所以，第0步的时候，每个顶点存储的是其自身的key信息以及初始邻近接壤节点信息。要访问源顶点需要使用_get(“someAttribute”)方法：

[js] view plaincopy

    switch(global.step) {  
     case 0:  
       result = {  
         value: vertex._get("_key"),  
         backwards: []  
       }  
       sendSender = true;  
       break;  
    }  

一个顶点只能访问其外部边界，因此在第1步的时候要记得把它所有接收到的消息放入数组中，以便进行向后通信，同时要根据传入的消息来更新节点组。

[js] view plaincopy

    case 1:  
      result = vertex._getResult();  
      while (message.hasNext()) {  
        next = message.next();  
        if (result.value < next.data) {  
          result.value = next.data;  
        }  
        result.backwards.push(next.sender);  
      }  
      break;  

所以前两步的操作开启了向前和向后通信，接着执行算法直到每个顶贴都接收到其顶点组标识信息。因此，当接收到邻近标识符信息后，每个顶点需要更新顶点组标识信息：

[js] view plaincopy

    default:  
      result = vertex._getResult();  
      while (message.hasNext()) {  
        next = message.next();  
        if (result.value < next.data) {  
          result.value = next.data;  
          changed = true;  
        }  
      }  

当一个顶点不再接收到新的消息或新的组标识时，要使它暂时失效。仅当再从邻近顶点接收新消息的时候进行激活：

[js] view plaincopy

    if (global.step > 1 && ! changed) {  
     vertex._deactivate();  
     return;  
    }  

如果接收到新的组标识时要把结果进行存储：

[js] view plaincopy

    vertex._setResult(result);  

接着要通知邻近顶点，包括向前与向后：

[js] view plaincopy

    while (vertex._outEdges.hasNext()) {  
      edge = vertex._outEdges.next();  
      message.sendTo(edge._getTarget(), result.value, sendSender);  
    }  

[js] view plaincopy

    for (var i = 0; i < result.backwards.length; ++i) {  
      message.sendTo(result.backwards, result.value, sendSender);  
    }  

然后失效该顶点直到接收到新的消息：

[js] view plaincopy

    vertex._deactivate();  

合成算法式 http://51python.lofter.com/post/198ab6_3e5be66

为了减少冗余的消息使得工作者算法更加高效，。比方说在该示例中，德国节点可能会收到来自奥地利和瑞士的消息；由于按字母排序，奥地利的消息可以忽略，从而减少不必要的消息接收。在Pregel中的消息合成器可定义为：Python3.4模拟实现生产者消费者模式