摘要：關閉套接字和上下文備注說明如何利用使用首先下載所需的包，解壓以后將和文件放到自己電腦中的安裝路徑中的文件夾下，最后需要將之前解壓后的包放在項目的中或者資源下載鏈接密碼項目源碼下載鏈接鏈接密碼

在講ZeroMQ前先給大家講一下什么是消息隊列。

消息隊列中間件是分布式系統中重要的組件，主要解決應用耦合，異步消息，流量削鋒等問題。實現高性能，高可用，可伸縮和最終一致性架構。是大型分布式系統不可缺少的中間件。目前在生產環境，使用較多的消息隊列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等。其實簡單點說，消息隊列就是如何使各分載器如何實現負載均衡使得完成分布式目標。

ZeroMQ是一種基于消息隊列的多線程網絡庫，其對套接字類型、連接處理、幀、甚至路由的底層細節進行抽象，提供跨越多種傳輸協議的套接字。ZeroMQ是網絡通信中新的一層，介于應用層和傳輸層之間（按照TCP/IP劃分），其是一個可伸縮層，可并行運行，分散在分布式系統間。ZeroMQ幾乎所有的I/O操作都是異步的，主線程不會被阻塞。ZeroMQ會根據用戶調用zmq_init函數時傳入的接口參數，創建對應數量的I/O Thread。每個I/O Thread都有與之綁定的Poller，Poller采用經典的Reactor模式實現，Poller根據不同操作系統平臺使用不同的網絡I/O模型（select、poll、epoll、devpoll、kequeue等）。主線程與I/O線程通過Mail Box傳遞消息來進行通信。Server開始監聽或者Client發起連接時，在主線程中創建zmq_connecter或zmq_listener，通過Mail Box發消息的形式將其綁定到I/O線程，I/O線程會把zmq_connecter或zmq_listener添加到Poller中用以偵聽讀/寫事件。Server與Client在第一次通信時，會創建zmq_init來發送identity，用以進行認證。認證結束后，雙方會為此次連接創建Session，以后雙方就通過Session進行通信。每個Session都會關聯到相應的讀/寫管道， 主線程收發消息只是分別從管道中讀/寫數據。Session并不實際跟kernel交換I/O數據，而是通過plugin到Session中的Engine來與kernel交換I/O數據。

【1】Request-Response

由請求端發起請求，然后等待回應端應答。一個請求必須對應一個回應，從請求端的角度來看是發-收配對，從回應端的角度是收-發對。跟一對一結對模型的區別在于請求端可以是1~N個。該模型主要用于遠程調用及任務分配等。Echo服務就是這種經典模型的應用。

下面通過Java實現這一模型：

server port

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class Server {

   public static void main(String[] args) throws InterruptedException {
       
       //實現服務器端的上下文及套接字
       Context context = ZMQ.context(1);
       Socket responder = context.socket(ZMQ.REP);
       
       //使服務器端通過tcp協議通信，監聽5555端口
       responder.bind("tcp://*:5555");
       while (!Thread.currentThread().isInterrupted()) {
           byte[] request = responder.recv(0);
           System.out.println("Received Hello");
           Thread.sleep(1000);
           String reply = "World";
           responder.send(reply.getBytes(), 0);
       }
       
       //關閉服務器端的上下文及套接字
       responder.close();
       context.close();
   }

}

client port

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class Client {

   public static void main(String[] args) {    
       //創立客戶端的上下文捷套接字
       Context context = ZMQ.context(1);
       System.out.println("Connecting to hello world server…");
       Socket requester = context.socket(ZMQ.REQ);
       
       //講客戶端綁定在5555端口
       requester.connect("tcp://localhost:5555");
          for (int requestNbr = 0; requestNbr != 100; requestNbr++) {
              String request = "Hello";
              System.out.println("Sending Hello " + requestNbr);
              requester.send(request.getBytes(), 0);
              byte[] reply = requester.recv(0);
              System.out.println("Received " + new String(reply) + " " + requestNbr);
          }      
          //關閉客戶端的上下文套接字
          requester.close();
          context.term();
   }

}

【2】Publisher/Subscriber model

發布端單向分發數據，且不關心是否把全部信息發送給訂閱端。如果發布端開始發布信息時，訂閱端尚未連接上來，則這些信息會被直接丟棄。訂閱端未連接導致信息丟失的問題，可以通過與請求回應模型組合來解決。訂閱端只負責接收，而不能反饋，且在訂閱端消費速度慢于發布端的情況下，會在訂閱端堆積數據。該模型主要用于數據分發。天氣預報、微博明星粉絲可以應用這種經典模型。

Server Port

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class ZMQ_PUB {

   public static void main(String[] args) throws InterruptedException {
       Context context = ZMQ.context(1);
       Socket publisher = context.socket(ZMQ.PUB);
       publisher.bind("tcp://*:5555");
       Thread.sleep(3000);
       for(int i=0;i<100;i++){
           publisher.send(("admin " + i).getBytes(), ZMQ.NOBLOCK);
           System.out.println("pub msg " + i);  
           Thread.sleep(1000);  
       }
       context.close();
       publisher.close();
   }

}

Client Port

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class ZMQ_SUB {

    public static void main(String[] args) {
        Context context = ZMQ.context(1);
        Socket subscriber = context.socket(ZMQ.SUB);
        subscriber.connect("tcp://localhost:5555");
         subscriber.subscribe("".getBytes());  
         for (int i=0;i<100;i++) {  
             //Receive a message.       
             String string = new String(subscriber.recv(0));           
             System.out.println("recv 1" + string);  
         }       
         //關閉套接字和上下文  
         subscriber.close();  
         context.term();  
    }
}

【3】push/pull

push port
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class Push {

    public static void main(String[] args) {
        Context context = ZMQ.context(1);
        Socket push = context.socket(ZMQ.PUSH);
        push.bind("ipc://fjs");
        for (int i = 0; i < 10000000; i++) {  
            push.send("hello".getBytes(), i);  
        }  
        push.close();  
        context.term();
    }
}

pull port
import java.util.concurrent.atomic.AtomicInteger;

import org.zeromq.ZMQ;

public class Pull {

public static void main(String args[]) {  
    final AtomicInteger number = new AtomicInteger(0);  
    for (int i = 0; i < 5; i++) {  
        new Thread(new Runnable(){  
            private int here = 0;  
            public void run() {  
                // TODO Auto-generated method stub  
                ZMQ.Context context = ZMQ.context(1);  
                ZMQ.Socket pull = context.socket(ZMQ.PULL);  
                pull.connect("ipc://fjs");  
                //pull.connect("ipc://fjs");  
                while (true) {  
                    String message = new String(pull.recv());  
                    int now = number.incrementAndGet();  
                    here++;  
                    if (now % 1000000 == 0) {  
                        System.out.println(now + "  here is : " + here);  
                    }  
                }  
            }                
        }).start();               
    }  
}  

}

備注說明：

【1】如何利用Java使用ZeroMQ
首先下載zmq所需的zip包，解壓以后將libzmq.dll和jzmq.dll文件放到自己電腦中的jdk安裝路徑中的bin文件夾下，最后需要將之前解壓后的zmq.jar包放在項目的lib中或者

zeromq資源下載：
鏈接：http://pan.baidu.com/s/1miuvSfQ 密碼：ttss

項目源碼下載鏈接：
鏈接：http://pan.baidu.com/s/1dE5Plr7 密碼：vqze