ZeroMQ--使用jzmq进行编程

标签： zeromq jzmq 编程 | 发表时间：2014-07-25 17:50 | 作者：kavy
出处：http://www.iteye.com

http://my.oschina.net/cloudcoder/blog/200989

 

一、环境搭建

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
wget http://download.zeromq.org/zeromq-2.1.7.tar.gz
tar -xzf zeromq-2.1.7.tar.gz
cd zeromq-2.1.7
./configure
make
sudo make install
 
git clone https://github.com/nathanmarz/jzmq.git
cd jzmq
./autogen.sh
./configure
make
sudo make install
 
如果没有安装libtool、libuuid-devel则需要先安装，否则安装失败
yum install libtool
yum install libuuid-devel

常见问题：

出现java.lang.UnsatisfiedLinkError: /usr/local/lib/libjzmq.so.0.0.0: libzmq.so.1: cannot open shared object file: No such file or directory异常 
原因是未找到zmq动态链接库。 
解决方法1：export LD_LIBRARY_PATH=/usr/local/lib 
解决方法2：编辑/etc/ld.so.conf文件，增加一行：/usr/local/lib。再执行sudo ldconfig命令 

Exception in thread "main" java.lang.UnsatisfiedLinkError: no jzmq in java.library.path 
未设置native library 
在eclipse设置native library为/usr/local/lib 
或在jvm增加参数 
-Djava.library.path=/usr/local/lib 
或在启动脚本中增加 
java -Djava.library.path=/usr/local/lib

二、使用jzmq进行编程

    1.创建maven项目，pom.xml的内容参见pom.xml

      注意：jzmq的版本不能太高，建议使用2.1.0，目前storm也是使用这个版本的jzmq-2.1.0.jar

      否则报： java.lang.UnsatisfiedLinkError: org.zeromq.ZMQ$Socket.nativeInit()V

    2.编写Publisher.java,Subscriber.java,参见源代码

    Publisher.java  

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package com.catt.mqtest.pubsub;
 
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;
 
public class Publisher {
 
     // 等待10个订阅者
     private static final int SUBSCRIBERS_EXPECTED = 10 ;
     // 定义一个全局的记录器，通过LoggerFactory获取
     private final static Logger log = LoggerFactory.getLogger(Publisher. class );
 
     public static void main(String[] args) throws InterruptedException{
         Context context = ZMQ.context( 1 );
         Socket publisher = context.socket(ZMQ.PUB);
         publisher.bind( "tcp://*:5557" );
         try {
             // zmq发送速度太快，在订阅者尚未与发布者建立联系时，已经开始了数据发布
             Thread.sleep( 1000 );
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
 
         publisher.send( "send start......" .getBytes(), 0 );
         for ( int i = 0 ; i < 10 ; i++) {
             publisher.send(( "Hello world " +i).getBytes(), ZMQ.NOBLOCK);
         }
         publisher.send( "send end......" .getBytes(), 0 );
 
         publisher.close();
         context.term();
     }
}

Subscriber.java

 

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
package com.catt.mqtest.pubsub;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;
 
public class Subscriber {
 
     // 定义一个全局的记录器，通过LoggerFactory获取
     private final static Logger log = LoggerFactory.getLogger(Subscriber. class );
 
     public static void main(String[] args) {
         Context context = ZMQ.context( 1 );
         Socket subscriber = context.socket(ZMQ.SUB);
         subscriber.connect( "tcp://192.168.230.128:5557" );
         subscriber.subscribe( "" .getBytes());
         int total = 0 ;
         while ( true ) {
             byte [] stringValue = subscriber.recv( 0 );
             String string = new String(stringValue);
             if (string.equals( "send end......" )) {
                 break ;
             }
             total++;
             System.out.println( "Received " + total + " updates. :" + string);
         }
 
         subscriber.close();
         context.term();
     }
}

pom.xml

 

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
< project xmlns = "http://maven.apache.org/POM/4.0.0" xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation = "http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" >
     < modelVersion >4.0.0</ modelVersion >
 
     < groupId >com.catt</ groupId >
     < artifactId >mqtest</ artifactId >
     < version >0.0.1-SNAPSHOT</ version >
     < packaging >jar</ packaging >
 
     < name >mqtest</ name >
     < url >http://maven.apache.org</ url >
 
     < properties >
         < project.build.sourceEncoding >UTF-8</ project.build.sourceEncoding >
     </ properties >
 
     < dependencies >
         < dependency >
             < groupId >org.zeromq</ groupId >
             < artifactId >jzmq</ artifactId >
             < version >2.1.0</ version >
         </ dependency >
         < dependency >
             < groupId >ch.qos.logback</ groupId >
             < artifactId >logback-classic</ artifactId >
             < version >1.1.1</ version >
         </ dependency >
         < dependency >
             < groupId >junit</ groupId >
             < artifactId >junit</ artifactId >
             < version >4.10</ version >
             < scope >test</ scope >
         </ dependency >
     </ dependencies >
</ project >


已有 0 人发表留言，猛击->> 这里<<-参与讨论


ITeye推荐



相关 [zeromq jzmq 编程] 推荐：

ZeroMQ--使用jzmq进行编程

- - 编程语言 - ITeye博客
如果没有安装libtool、libuuid-devel则需要先安装，否则安装失败. 出现java.lang.UnsatisfiedLinkError: /usr/local/lib/libjzmq.so.0.0.0: libzmq.so.1: cannot open shared object file: No such file or directory异常 .

ZeroMQ(java)中监控Socket

- - CSDN博客架构设计推荐文章
基本上ZeroMQ（java）中基本的代码都算是过了一遍了吧，不过觉得它在日志这一块貌似基本没有做什么工作，也就是我们通过日志来知道ZeroMQ都发生了什么事情. 而且由于ZeroMQ中将连接的建立和重连接都进行了隔离，用户不需要做什么事情来维护连接，当然这样做的好处是使程序员的编码工作变少了，但是当然也有不好的地方，那就是用户失去了对ZeroMQ整个运行阶段的控制.

[转][转]ZeroMQ 深度探索

- - heiyeluren的blog（黑夜路人的开源世界）
最初认识 ZeroMQ 是被它的名号所吸引，最近在一个高性能中间件的项目中用到了 ZeroMQ，对这个号称“史上最快的消息队列”有了更深层次的了解. 如果我们仅仅把 ZeroMQ 看作是一个消息队列，那就完全搞错了，ZeroMQ 是一套智能传输层协议，它不仅为开发者提供了强大的开发包，还包含了一套很棒的通信协议的实现，更值得一提是，它对分布式系统开发有着相当独到的见解，绝对值得我们好好学习.

[转]史上最快消息内核——ZeroMQ

- 乾 - heiyeluren的blog（黑夜路人的开源世界）
来源：http://blog.dccmx.com/2011/02/zeromq/ . ZeroMQ是一个很有个性的项目，它原来是定位为“史上最快消息队列”，所以名字里面有“MQ”两个字母，但是后来逐渐演变发展，慢慢淡化了消息队列的身影，改称为消息内核，或者消息层了. 从网络通信的角度看，它处于会话层之上，应用层之下，有了它，你甚至不需要自己写一行的socket函数调用就能完成复杂的网络通信工作.

ZeroMQ(java)入门之Requerst/Response模式

- - CSDN博客架构设计推荐文章
自己最开始是在cloud foundry中接触过消息服务器（nats），或者说是消息中间件，也算是初步知道了一个消息服务器对于分布式的网络系统的重要性，后来自己也曾想过在一些项目中使用它，尤其是在一些分布式的环境下，可以极大的方便整个系统的实现. 在中间通过一个消息中间件，可以很方便的将各个woker的数据发送到最终的统计服务器来做数据的统计，从而减少很多在网络通信中会消耗的编程时间.

zmq-rpc:基于zeromq网络层编写的protobuf RPC框架

- Shengbin - codedump
阅读过zmq的代码之后,感觉这个网络层是我目前见过最高效的–线程之间使用lockfree的消息队列保存消息,可以启动多个I/O线程分担压力等等特性.于是决定基于它写一个protobuf RPC的框架.. 另外,这里使用的protobuf是旧版本2.3.0,新版本2.4.1的生成的RPC service接口跟原来不太一致,暂时还没有去研究它.BTW,升级版本之后导致原来的接口发生变化这是一个很操蛋的事情..

为什么我希望用C而不是C++来实现ZeroMQ

- - 博客 - 伯乐在线
开始前我要先做个澄清：这篇文章同Linus Torvalds这种死忠C 程序员吐槽C++的观点是不同的. 在我的整个职业生涯里我都在使用C++，而且现在C++依然是我做大多数项目时的首选 编程语言. 自然的，当我从2007年开始做 ZeroMQ( ZeroMQ项目主页)时，我选择用C++来实现.

消息中间件的技术选型心得－RabbitMQ、ActiveMQ和ZeroMQ

- - haohtml's blog
RabbitMQ、ActiveMQ和ZeroMQ都是极好的消息中间件，但是我们在项目中该选择哪个更适合呢. 下面我会对这三个消息中间件做一个比较，看了后你们就心中有数了. RabbitMQ是AMQP协议领先的一个实现，它实现了代理(Broker)架构，意味着消息在发送到客户端之前可以在中央节点上排队.

Hadoop Streaming 编程

- - 学着站在巨人的肩膀上
Hadoop Streaming是Hadoop提供的一个编程工具，它允许用户使用任何可执行文件或者脚本文件作为Mapper和Reducer，例如：. 采用shell脚本语言中的一些命令作为mapper和reducer（cat作为mapper，wc作为reducer）. 本文安排如下，第二节介绍Hadoop Streaming的原理，第三节介绍Hadoop Streaming的使用方法，第四节介绍Hadoop Streaming的程序编写方法，在这一节中，用C++、C、shell脚本 和python实现了WordCount作业，第五节总结了常见的问题.

Shell编程

- - 博客园_首页
本来打算寒假回家好好学习Linux的，为以后学习嵌入式打好基础的. 回家之后的学习效率非常低，之前为了搭建Linux环境，折腾了很长时间，学到现在也就勉强才把Shell编程学完了. 今天就把自己学习的相关知识点总结整理一下. 个人感觉shell程序跟windows下的批处理文件有点像，就是将一些系统命令写进一个可执行文件中，然后执行.