Java调用ZMQ

经过上面两篇博客已经对ZMQ和如何编译ZMQ有所了解了，在第一篇文章中介绍了ZMQ有三种模式，这篇博客就写一个关于发布订阅模式的Demo，希望对读者能够起到学习的作用，那要是在深入学习的话，可以参照zmq的官方和它自身带的实例代码学习。

 

在写DEMO之前需要将上篇博客编译之后的lib文件配置到环境变量中，

C:\Users\CJQ\Desktop\zeromq-2.2.0\lib;C:\Users\CJQ\Desktop\zeromq-2.2.0\zeromq-jzmq\lib

也就是将zmq的lib文件夹和jzmq的文件夹放到环境变量中。

 

之后我们来写发布端代码，

 

[java] view plaincopyprint?

1. package com.wh.mq.demo2;  
2.    
3. import org.zeromq.ZMQ;  
4. import org.zeromq.ZMQ.Context;  
5. import org.zeromq.ZMQ.Socket;  
6.    
7. public class Sync_PUB {  
8.    
9.    public static void main(String[] args) throws InterruptedException {  
10.        Context context = ZMQ.context(1);  
11.        Socket publisher = context.socket(ZMQ.PUB);  
12.        publisher.bind("tcp://*:5561");  
13.        //zmq发送速度太快，在订阅者尚未与发布者建立联系时，已经开始了数据发布  
14.        Thread.sleep(1000);  
15.    
16.        int update_nbr;  
17.          
18.        for (update_nbr = 20; update_nbr < 40; update_nbr++) {  
19.                 Stringa="{\"magicNum\":\"CHINSOFT\",\"varName\":\"ZJ_YD_1\",\"varType\":\"5\",\"varValue\":"+update_nbr+",\"varQuality\":\"1111\",\"varTime\":"+System.currentTimeMillis()/1000+"}";  
20.            publisher.send(a.getBytes(), ZMQ.NOBLOCK);  
21.            System.out.println(update_nbr);  
22.            Thread.sleep(1000);  
23.        }  
24.          
25.    
26.        publisher.close();  
27.        context.term();  
28.     }  
29. }  

 

发布端需要通过context.socket（ZMQ.PUB）表示为发布端，通过bind方法来创建发布端连接，等待订阅者连接。

之后通过send方法将数据发送到出去。

 

之后来写订阅端代码

[java] view plaincopyprint?

1. public classSync_SUB1 {  
2.    
3.    publicstaticvoidmain(String[] args) {  
4.         Context context = ZMQ.context(1);  
5.         Socket subscriber = context.socket(ZMQ.SUB);  
6.         subscriber.connect("tcp://localhost:5561");  
7.    
8.         //设置订阅条件"setsockopt"  
9.         subscriber.subscribe("".getBytes());  
10.         int update_nbr = 0;  
11.         while (true) {  
12.             byte[] stringValue = subscriber.recv(0);  
13.    
14.             String string = new String(stringValue);  
15.            
16.             update_nbr++;  
17.             System.out.println("Received " + update_nbr + " updates. :"+ string);  
18.         }  
19.    }  
20. }  

 

客户端通过connect进行连接，之后通过recv来进行数据接收。

 

到此为止发布订阅Demo就写完了，通过这三篇博客能够对ZMQ有了初步的认识和简单实用，希望这三篇博客对学习zmq的读者有所帮助。