ZeroMQ研究 Majordomo Protocol， Java樣例實現

最近研究利用zeromq實現多對多的雙向自由收發。在官方上發現了MDP協議，經過驗證貌似可行。正在開發中，將驗證程式碼分享如下。

互動協議棧： Worker端：

A READY command consists of a multipart message of 4 frames, formatted on the wire as follows: * Frame 0: Empty frame * Frame 1: "MDPW01" (six bytes, representing MDP/Worker v0.1) * Frame 2: 0x01 (one byte, representing READY) * Frame 3: Service name (printable string) A REQUEST command consists of a multipart message of 6 or more frames, formatted on the wire as follows: * Frame 0: Empty frame * Frame 1: "MDPW01" (six bytes, representing MDP/Worker v0.1) * Frame 2: 0x02 (one byte, representing REQUEST) * Frame 3: Client address (envelope stack) * Frame 4: Empty (zero bytes, envelope delimiter) * Frames 5+: Request body (opaque binary) A REPLY command consists of a multipart message of 6 or more frames, formatted on the wire as follows: * Frame 0: Empty frame * Frame 1: "MDPW01" (six bytes, representing MDP/Worker v0.1) * Frame 2: 0x03 (one byte, representing REPLY) * Frame 3: Client address (envelope stack) * Frame 4: Empty (zero bytes, envelope delimiter) * Frames 5+: Reply body (opaque binary) A HEARTBEAT command consists of a multipart message of 3 frames, formatted on the wire as follows: * Frame 0: Empty frame * Frame 1: "MDPW01" (six bytes, representing MDP/Worker v0.1) * Frame 2: 0x04 (one byte, representing HEARTBEAT) A DISCONNECT command consists of a multipart message of 3 frames, formatted on the wire as follows: * Frame 0: Empty frame * Frame 1: "MDPW01" (six bytes, representing MDP/Worker v0.1) * Frame 2: 0x05 (one byte, representing DISCONNECT)

Client端：

A REQUEST command consists of a multipart message of 4 or more frames, formatted on the wire as follows: * Frame 0: Empty (zero bytes, invisible to REQ application) * Frame 1: "MDPC01" (six bytes, representing MDP/Client v0.1) * Frame 2: Service name (printable string) * Frames 3+: Request body (opaque binary) A REPLY command consists of a multipart message of 4 or more frames, formatted on the wire as follows: * Frame 0: Empty (zero bytes, invisible to REQ application) * Frame 1: "MDPC01" (six bytes, representing MDP/Client v0.1) * Frame 2: Service name (printable string) * Frames 3+: Reply body (opaque binary)

下面是示例程式碼，基於官方的程式碼精簡改造，去掉了heartbeat機制。便於理解功能。 Broker：

package com.coderli.zeromq.majordomoprotocol;


import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.Map;


import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg 
;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Majordomo Protocol協議驗證<br>
 * 用於實現多client、多worker實現雙向指定目標資料收發 <br>
 * 此為核心Broker模組
 *
 */
public class Broker extends JZMQBase {


     private static class Service {
           // 服務名
           public final String name;
           // 請求資訊佇列
          Deque<ZMsg> requests;
           // 待用worker佇列
          Deque<Worker> waiting; // List of waiting workers


           public Service(String name) {
               this. name = name;
               this. requests = new ArrayDeque<ZMsg>();
               this. waiting = new ArrayDeque<Worker>();
          }
     }


     private static class Worker {
           // worker的唯一標識
           @SuppressWarnings( "unused")
          String identity; // Identity of worker
           // 目標worker地址
          ZFrame address; // Address frame to route to
           // 包含的service名稱，如果存在
          Service service;


           public Worker(String identity, ZFrame address) {
               this. address = address;
               this. identity = identity;
          }
     }


     private ZContext ctx;
     private ZMQ.Socket socket;
     private Map<String, Service> services;
     private Map<String, Worker> workers;
     private Deque<Worker> waiting;


     public static void main(String[] args) {
          Broker broker = new Broker();
          broker.bind( BROKER_FRONT_END);
          broker.mediate();
     }


     public Broker() {
           this. services = new HashMap<String, Service>();
           this. workers = new HashMap<String, Worker>();
           this. waiting = new ArrayDeque<Worker>();
           this. ctx = new ZContext();
           this. socket = ctx.createSocket(ZMQ. ROUTER);
     }


     public void mediate() {
           while (!Thread. currentThread().isInterrupted()) {
              ZMQ.Poller items = new ZMQ.Poller(1);
              items.register( socket, ZMQ.Poller. POLLIN);
              items.poll();
               if (items.pollin(0)) {
                   ZMsg msg = ZMsg. recvMsg(socket);
                    if (msg == null) {
                        System. out.println( "接收到的訊息為null。" );
                         break; // Interrupted
                   }
                   System. out.println( "I: received message:\n");
                   msg.dump(System. out);
                    // 根據協議棧規則讀取資料，此處需要注意broker接受到的協議棧格式
                   ZFrame sender = msg.pop();
                   ZFrame empty = msg.pop();
                   ZFrame header = msg.pop();
                    if (MDP. C_CLIENT.frameEquals(header)) {
                        processClient(sender, msg);
                   } else if (MDP.W_WORKER.frameEquals(header))
                        processWorker(sender, msg);
                    else {
                        System. out.println( "E: invalid message:\n");
                        msg.dump(System. out);
                        msg.destroy();
                   }
                   sender.destroy();
                   empty.destroy();
                   header.destroy();
              }
          }
          destroy();
     }


     private void destroy() {
          Worker[] deleteList = workers.entrySet().toArray( new Worker[0]);
           for (Worker worker : deleteList) {
              deleteWorker(worker, true);
          }
           ctx.destroy();
     }


     /**
      * 處理客戶端請求的，用於分發給指定的worker.
      */
     private void processClient(ZFrame sender, ZMsg msg) {
           if (msg.size() < 2) {
              System. out.println( "訊息棧不完整，不能傳送" );
               return;
          }
          ZFrame serviceFrame = msg.pop();
          msg.wrap(sender);
          dispatch(requireService(serviceFrame), msg);
          serviceFrame.destroy();
     }


     private void processWorker(ZFrame sender, ZMsg msg) {
           if (msg.size() < 1) {
              System. out.println( "回覆給客戶端的訊息不完整，不能傳送。" );
          }
          ZFrame command = msg.pop();
           boolean workerReady = workers.containsKey(sender.strhex());
          Worker worker = requireWorker(sender);
           if (MDP. W_READY.frameEquals(command)) {
               if (workerReady) {
                   System. out.println( "刪除worker：" + sender.strhex());
                   deleteWorker(worker, true);
              } else {
                   ZFrame serviceFrame = msg.pop();
                   worker. service = requireService(serviceFrame);
                   workerWaiting(worker);
                   serviceFrame.destroy();
              }
          } else if (MDP. W_REPLY.frameEquals(command)) {
               if (workerReady) {
                   System. out.println( "開始給客戶端相應" );
                   ZFrame client = msg.unwrap();
                   msg.addFirst(worker. service. name);
                   msg.addFirst(MDP. C_CLIENT.newFrame());
                   msg.wrap(client);
                   msg.send( socket);
                   workerWaiting(worker);
              } else {
                   deleteWorker(worker, true);
              }
          } else {
              System. out.print( "不合法的訊息結構" );
              msg.dump(System. out);
          }
          msg.destroy();
     }


     private void deleteWorker(Worker worker, boolean disconnect) {
          System. out.println( "刪除worker");
           if (disconnect) {
              sendToWorker(worker, MDP. W_DISCONNECT, null, null);
          }
           if (worker. service != null)
              worker. service. waiting.remove(worker);
           workers.remove(worker);
          worker. address.destroy();
     }


     private Worker requireWorker(ZFrame address) {
           assert (address != null);
          String identity = address.strhex();
          Worker worker = workers.get(identity);
           if (worker == null) {
              worker = new Worker(identity, address.duplicate());
               workers.put(identity, worker);
              System. out.println( "註冊了新的worker：" + identity);
          }
           return worker;
     }


     private Service requireService(ZFrame serviceFrame) {
           assert (serviceFrame != null);
          String name = serviceFrame.toString();
          Service service = services.get(name);
           if (service == null) {
              service = new Service(name);
               services.put(name, service);
          }
           return service;
     }


     private void bind(String endpoint) {
           socket.bind(endpoint);
          System. out.println( "Broker版定在埠： " + endpoint);
     }


     public synchronized void workerWaiting(Worker worker) {
           waiting.addLast(worker);
          worker. service. waiting.addLast(worker);
          dispatch(worker. service, null);
     }


     private void dispatch(Service service, ZMsg msg) {
           assert (service != null);
           if (msg != null) {
              service. requests.offerLast(msg);
          }
           while (!service. waiting.isEmpty() &amp;&amp; !service.requests.isEmpty()) {
              msg = service. requests.pop();
              Worker worker = service. waiting.pop();
               waiting.remove(worker);
              sendToWorker(worker, MDP. W_REQUEST, null, msg);
              msg.destroy();
          }
     }


     public void sendToWorker(Worker worker, MDP command, String option,
              ZMsg msgp) {
          ZMsg msg = msgp == null ? new ZMsg() : msgp.duplicate();
           if (option != null)
              msg.addFirst( new ZFrame(option));
          msg.addFirst(command.newFrame());
          msg.addFirst(MDP. W_WORKER.newFrame());
          msg.wrap(worker. address.duplicate());
          System. out.println( "給worker傳送命令： [" + command + "]。");
          msg.dump(System. out);
          msg.send( socket);
     }
}

ClientAPI:

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;


/**
 * ZeroMQ Majordomo Protocol協議驗證<br>
 * 用於實現多client、多worker實現雙向指定目標資料收發 <br>
 * 此為Client端依賴的API。
 *
 */
public class ClientAPI {


     private String broker;
     private ZContext ctx;
     private ZMQ.Socket client;
     private long timeout = 2500;
     private int retries = 3;


     public long getTimeout() {
           return timeout;
     }


     public void setTimeout( long timeout) {
           this. timeout = timeout;
     }


     public int getRetries() {
           return retries;
     }


     public void setRetries( int retries) {
           this. retries = retries;
     }


     public ClientAPI(String broker) {
           this. broker = broker;
           ctx = new ZContext();
          reconnectToBroker();
     }


     void reconnectToBroker() {
           if ( client != null) {
               ctx.destroySocket( client);
          }
           client = ctx.createSocket(ZMQ. REQ);
           client.connect( broker);
          System. out.println( "連線到Broker：" + broker );
     }


     /**
      * 給broker傳送訊息
      *
      * @param service
      * @param request
      * @return
      */
     public ZMsg send(String service, ZMsg request) {


          request.push( new ZFrame(service));
          request.push(MDP. C_CLIENT.newFrame());
          System. out.println( "傳送訊息給worker：" + service);
          request.dump(System. out);
          ZMsg reply = null;


           int retriesLeft = retries;
           while (retriesLeft > 0 &amp;&amp; !Thread.currentThread().isInterrupted()) {
              request.duplicate().send( client);
              ZMQ.Poller items = new ZMQ.Poller(1);
              items.register( client, ZMQ.Poller. POLLIN);
               if (items.poll( timeout) == -1)
                    break; // 超時退出
               if (items.pollin(0)) {
                   ZMsg msg = ZMsg. recvMsg(client);
                   System. out.println( "接收到訊息。" );
                   msg.dump(System. out);
                   ZFrame header = msg.pop();
                   header.destroy();
                   ZFrame replyService = msg.pop();
                   replyService.destroy();
                   reply = msg;
                    break;
              } else {
                   items.unregister( client);
                    if (--retriesLeft == 0) {
                        System. out.println( "超過重試次數，錯誤。退出。" );
                         break;
                   }
                   System. out.println( "沒有收到迴應，重試。" );
                   reconnectToBroker();
              }
          }
          request.destroy();
           return reply;
     }


     public void destroy() {
           ctx.destroy();
     }
}

ClientOne:

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZMsg;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Majordomo Protocol協議驗證<br>
 * 用於實現多client、多worker實現雙向指定目標資料收發 <br>
 * 此為Client端一號，定向發給1、2號worker
 *
 */
public class ClientOne extends JZMQBase {


     public static void main(String[] args) throws InterruptedException {
          ClientAPI clientSession = new ClientAPI(BROKER_FRONT_END);


           int count;
           for (count = 0; count < 1; count++) {
              ZMsg request = new ZMsg();
              ZMsg reply = null;
               long start = System. nanoTime();
              request.addString(String. valueOf(start));
               if (count % 2 == 1) {
                   reply = clientSession.send( "one", request);
              } else {
                   reply = clientSession.send( "two", request);
              }
               if (reply != null)
                   reply.destroy();
               else
                    break; // Interrupt or failure
              Thread. sleep(1000000L);
          }


          System. out.printf( "%d requests/replies processed\n", count);
          clientSession.destroy();
     }
}

WorkerAPI:

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;


/**
 * ZeroMQ驗證 workerAPI封裝
 *
 * @author lihzh
 * @date 2014年1月15日 下午2:23:14
 */
public class WorkerAPI {


     private String broker;
     private ZContext ctx;
     private String service;
     private ZMQ.Socket worker;


     private long timeout = 2500;


     private ZFrame replyTo;


     public WorkerAPI(String broker, String service) {
           assert (broker != null);
           assert (service != null);
           this. broker = broker;
           this. service = service;
           ctx = new ZContext();
          reconnectToBroker();
     }


     /**
      * 給Broker傳送訊息
      *
      * @param command
      * @param option
      * @param msg
      */
     void sendToBroker(MDP command, String option, ZMsg msg) {
          msg = msg != null ? msg.duplicate() : new ZMsg();


           if (option != null)
              msg.addFirst( new ZFrame(option));
          msg.addFirst(command.newFrame());
          msg.addFirst(MDP. W_WORKER.newFrame());
          msg.addFirst( new ZFrame( new byte[0]));
          msg.send( worker);
     }


     void reconnectToBroker() {
           if ( worker != null) {
               ctx.destroySocket( worker);
          }
           worker = ctx.createSocket(ZMQ. DEALER);
           worker.connect( broker);
          sendToBroker(MDP. W_READY, service, null);
     }


     /**
      * 接收資料
      *
      * @param reply
      * @return
      * @author lihzh
      * @date 2014年1月15日 下午2:24:23
      */
     public ZMsg receive(ZMsg reply) 
 
              
           
              
              
            
            相關推薦
			   
            
            
            
 

    

    
    ZeroMQ研究 Majordomo Protocol， Java樣例實現
      

      
      

      
        
	最近研究利用zeromq實現多對多的雙向自由收發。在官方上發現了MDP協議，經過驗證貌似可行。正在開發中，將驗證程式碼分享如下。


	


	互動協議棧：


	Worker端：

	
		A READY command consists  

  
 

    

    
    【轉載】對一致性Hash算法，Java代碼實現的深入研究
      困難   之前   存在   itl   ger   正常   我不   操作   算法實現   原文地址：http://www.cnblogs.com/xrq730/p/5186728.html
 

一致性Hash算法
關於一致性Hash算法，在我之前的博文中已經有多次提到了，MemCache超詳細解讀一 

  
 

    

    
    對一致性Hash算法，Java代碼實現的深入研究
      memcach   還原   情況   ()   實用   target   強人   最壞情況   一致性hash   
一致性Hash算法
關於一致性Hash算法，在我之前的博文中已經有多次提到了，MemCache超詳細解讀一文中"一致性Hash算法"部分，對於為什麽要使用一致性Hash算法、一致性Has 

  
 

    

    
    數據庫.MongoDB.Java樣例
      slist   rgs   oss   lease   執行   bson   下載列表   releases   ron   1.先在MongoDB官網下載Java驅動包
MongoDB Java Driver:
http://mongodb.github.io/mongo-java-driver/
 
J 

  
 

    

    
    一個可以在針式打印機上打印的Java樣例
      exceptio   s2d   str   接口   draw   void   exception   println   入口   代碼：
import java.awt.Color;
import java.awt.Font;
import java.awt.Graphics;
import java 

  
 

    

    
    Scala中的類和物件，實用樣例詳細解析！
      
							
							
							
Scala類
簡介
類是物件的抽象，而物件是類的具體例項。類是抽象的，不佔用記憶體，而物件是具體的，佔用儲存空間。類是用於建立物件的藍圖，它是一個定義包括在特定型別的物件中的方法和變數的軟體模板。
Scala中的類不宣告為public，一個Scala原始檔中可 

  
 

    

    
    highchart 柱狀圖，單個樣例
       
 var chart = Highcharts.chart('container', {
    chart: {
        type: 'column'
    },
    title: {
        text: '月平均氣溫'
    },
    subtitle: {
 

  
 

    

    
    RedisUtil工具類，Java單例模式實現
      
							
							
							基於Java單例模式和Jedis封裝的RedisUtil，提供了Redis中的Key、字串、Hash、ZSet等的操作。



package com.cumt.tools.redis;

import com.alibaba.fastjson.JSON;
im 

  
 

    

    
    網易新聞API介面的實現（Picasso類庫的使用與HTTP協議的get請求，BitMap樣例（未使用））
      
							
							
							



  實體類：


public class NewsList implements Serializable{
    private String ctime,title,description,picUrl,url;

    public News 

  
 

    

    
    不用任何框架，Java 就能實現定時任務的 3 種方法！
      是的，不用任何框架，用我們樸素的 Java 程式語言就能實現定時任務。

今天，棧長就介紹 3 種實現方法，教你如何使用 JDK 實現定時任務！

## 1、 sleep

這也是我們最常用的 sleep 休眠大法，不只是當作休眠用，我們還可以利用它很輕鬆的能實現一個簡單的定時任務。

**實現邏輯：**

 

  
 

    

    
    hbase java api樣例（版本1.3.1，新API）
      quorum   desc   color   -i   arraylist   byte   logs   sin   fig   驗證了如下幾種java api的使用方法。
1.創建表
2.創建表（預分區）
3.單條插入
4.批量插入
5.批量插入（寫緩存）
6.單條get
7.批量get
8.簡單sca 

  
 

    

    
    Java多執行緒-----執行緒池的使用，原理以及舉例實現（三）（四）：使用樣例及如何配置執行緒池大小
      
                三.使用示例

　　前面我們討論了關於執行緒池的實現原理，這一節我們來看一下它的具體使用：

public class Test {
     public static void main(String[] args) {   
         ThreadPoolExe 

  
 

    

    
    Java中BIO，NIO和AIO使用樣例
      
                


上文中分析了阻塞，非阻塞，同步和非同步概念上的區別以及各種IO模型的操作流程，本篇文章將主要介紹Java中BIO，NIO和AIO三種IO模型如何使用。需要注意的是，本文中所提到的所有樣例都是在一個server對應一個client的情況下工作的，如果你想擴充套件為一個se 

  
 

    

    
    Java線程演示樣例 - 繼承Thread類和實現Runnable接口
      無法   ads   rup   args   dex   rom   一個   nts   future   

    進程（Process）和線程（Thread）是程序執行的兩個基本單元。Java並發編程很多其它的是和線程相關。
進程
    進程是一個獨立的執行單元，可將其視為一個程序或應用。然而，一 

  
 

    

    
    Java 反射經常用法演示樣例
      declared   演示   art   dsm   red   cls   div   rgs   public   

<pre name="code" class="java">import java.lang.reflect.Constructor;
import java.lang.r 

  
 

    

    
    Java-CyclicBarrier的簡單樣例
      rri   popu   exceptio   dex   cut   bit   content   -a   sub   

內容：一個主任務等待兩個子任務，通過CyclicBarrier的await()實現。此Runnable任務在CyclicBarrier的數目達到後，全部其他線程被喚醒前被運行。
p 

  
 

    

    
    Java之旅hibernate（3）——第一個hibernate的樣例
      數據庫方言   自由   article   存儲   代理   mit   rup   方便   post   

　在解說樣例之前。我們首先來理解一下hibernate的工作原理。理解原理將會幫助我們更好地理解hibernate和運用hibernate。
　1.      原理圖
　　利用hibe 

  
 

    

    
    java基礎入門-多線程同步淺析-以銀行轉賬為樣例
      stat   是什麽   0.11   如何   人員   ()   簡單   ret   沒有   

在說之前先普及一下線程是什麽？
線程：說白了就是一個任務片段
進程：是一個具有獨立功能的程序關於某個數據集合的一次執行活動。一個進程有一個或者多個線程


線程與進程的本質差別就是有麽有數據 

  
 

    

    
    圖標插件--jqplot實現柱狀圖及餅圖，表盤圖演示樣例
      itl   nbsp   left   柱狀圖   draw   ted   ner   ann   name   

柱狀圖
在jqPlot圖表插件使用說明（一）中，我們已經能夠通過jqPlot繪制出比較簡單的線形圖。通過查看源碼。我們也能夠看出，線形圖是jqPlot默認的圖表類型：




 

  
 

    

    
    TreeSet排序，存儲自己定義對象，自己定義比較器演示樣例
      strlen   覆蓋   ()   num   imp   rgs   com   哈希   ash   

Set:無序。不能夠反復元素。
|--HashSet：數據結構是哈希表。線程是非同步的。
保證元素唯一性的原理：推斷元素的hashCode值是否同樣。
假設同樣，還會繼續推斷