jeroMQ操作ZMQ实例

ZMQ即zeroMQ,ZMQ并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，它更像是一个底层的网络通讯库，在Socket API之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的API接口。

ZMQ提供了三个基本的通信模型，分别是“Request-Reply “(请求响应)，”Publisher-Subscriber“(发布订阅)，”Parallel Pipeline(pull/push)”

jeromq 是对 zoremq 框架的一个纯粹的java版本的实现（不需要依赖zoremq 动态库）
jeromq 它是基于java nio 进行的封装。
maven 依赖，仅一个jar

<dependency>
	<groupId>org.zeromq</groupId>
	<artifactId>jeromq</artifactId>
	<version>0.5.2</version>
</dependency>

1.请求响应模式

需要注意的是：
a) 服务端和客户端无论谁先启动，效果是相同的，这点不同于Socket。
b) 在服务端收到信息以前，程序是阻塞的，会一直等待客户端连接上来。
c) 服务端收到信息以后，会send一个“World”给客户端。值得注意的是一定是client连接上来以后，send消息给Server，然后 Server再rev然后响应client，这种一问一答式的。如果Server先send，client先rev是会报错的。
d) ZMQ通信通信单元是消息，他除了知道Bytes的大小，他并不关心的消息格式。因此，你可以使用任何你觉得好用的数据格式。Xml、Protocol Buffers、Thrift、json等等。
e) 虽然可以使用ZMQ实现HTTP协议，但是，这绝不是他所擅长的。

服务端

package zoremq;
 
import org.zeromq.ZMQ;
 
public class Request {
	public static void main(String args[]) {
		for (int j = 0; j < 5; j++) {
			new Thread(new Runnable() {
				public void run() {
					ZMQ.Context context = ZMQ.context(1); // 创建一个I/O线程的上下文
					ZMQ.Socket socket = context.socket(ZMQ.REQ); // 创建一个request类型的socket，这里可以将其简单的理解为客户端，用于向response端发送数据
 
					socket.connect("tcp://127.0.0.1:5555"); // 与response端建立连接
					long now = System.currentTimeMillis();
					for (int i = 0; i < 100000; i++) {
						String request = "hello";
						socket.send(request); // 向reponse端发送数据
						byte[] response = socket.recv(); // 接收response发送回来的数据
															// 正在request/response模型中，send之后必须要recv之后才能继续send，这可能是为了保证整个request/response的流程走完
						System.out.println("receive : " + new String(response));
					}
					long after = System.currentTimeMillis();
 
					System.out.println((after - now) / 1000);
				}
 
			}).start();
			;
		}
 
	}
}

客户端:

package zoremq;
 
import org.zeromq.ZMQ;
 
public class Response {
	public static void main(String[] args) {
		ZMQ.Context context = ZMQ.context(1); // 这个表示创建用于一个I/O线程的context
 
		ZMQ.Socket socket = context.socket(ZMQ.REP); // 创建一个response类型的socket，他可以接收request发送过来的请求，其实可以将其简单的理解为服务端
		socket.bind("tcp://*:5555"); // 绑定端口
		int i = 0;
		int number = 0;
		while (!Thread.currentThread().isInterrupted()) {
			i++;
			if (i == 10000) {
				i = 0;
				System.out.println(++number);
			}
			byte[] request = socket.recv(); // 获取request发送过来的数据
			System.out.println("客户端的相应：" + new String(request));
			// System.out.println("receive : " + new String(request));
			String response = "world";
			socket.send(response); // 向request端发送数据
									// ，必须要要request端返回数据，没有返回就又recv，将会出错，这里可以理解为强制要求走完整个request/response流程
		}
		socket.close(); // 先关闭socket
		context.term(); // 关闭当前的上下文
 
	}
}

2.发布订阅模式

发送端

package zoremq.sub;
 
import org.zeromq.ZMQ; 
import org.zeromq.ZMQ.Context; 
import org.zeromq.ZMQ.Socket; 
 
public class Publisher { 
 
    /** 
     * @param args 
     * @throws InterruptedException 
     */ 
    public static void main(String[] args) throws InterruptedException { 
    // TODO Auto-generated method stub 
    Context context = ZMQ.context(1); 
    Socket publisher = context.socket(ZMQ.PUB); 
    publisher.bind("tcp://*:5557"); 
    int i = 0; 
    while (true) { 
        Thread.currentThread().sleep(1000); 
        publisher.send("A".getBytes(), ZMQ.SNDMORE); 
        publisher.send("This is A".getBytes(), 0); 
        publisher.send("B".getBytes(), ZMQ.SNDMORE); 
        publisher.send("This is B".getBytes(), 0); 
    } 
    } 
 
}

接收端

package zoremq.sub;
 
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;
 
public class Sub1 {
	public static void main(String[] args) {
		Context context = ZMQ.context(1);
		Socket subscribe = context.socket(ZMQ.SUB);
		subscribe.connect("tcp://127.0.0.1:5557");
		subscribe.subscribe("B".getBytes());
		while (true) {
			System.out.println(new String(subscribe.recv(0)));
			System.out.println(new String(subscribe.recv(0)));
		}
	}
 
}

3.管道模式(pull/push)

发送端

package zoremq.PipeLine;
 
import org.zeromq.ZMQ;
public class MainPusher {
	public static void main(String[] args) throws InterruptedException {
		// 参数代表使用多少线程，大多数情况下，1个线程已经足够。
		ZMQ.Context context = ZMQ.context(1);
		// 指定模式为Pusher
		ZMQ.Socket socket = context.socket(ZMQ.PUSH);
		socket.connect("tcp://127.0.0.1:5557"); // 绑定服务地址及端口,如果是远程发送则这里的IP应是接收端服务器的IP
		for (;;) {
			long time = System.nanoTime();
			socket.send(String.valueOf(time));
			System.out.println("发布了新消息，时间：" + time);
			Thread.sleep(2000);
		}
	}
}

接收端

package zoremq.PipeLine;
import org.zeromq.ZMQ;
 
public class Puller   {
     public static void main(String[] args) {
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为pull模式
          ZMQ.Socket receiver = context.socket(ZMQ. PULL);
          receiver.bind("tcp://127.0.0.1:5557");
           for (;;) {
               byte[] recs = receiver.recv();
               long receiveTime = System.nanoTime();
              String oriMsg = new String(recs);
              String msg = new String(recs,1,recs.length-1);
               long pubTime = Long. valueOf(msg);
               long costTime = receiveTime - pubTime;
              System. out.println( "Receive: " + oriMsg + " Cost time: " + costTime);
          }
     }
}