jzmq消息发布订阅

Pom文件

	<dependencies>
		<dependency>
			<groupId>ch.qos.logback</groupId>
			<artifactId>logback-classic</artifactId>
			<version>1.1.1</version>
		</dependency>
		<dependency>
			<groupId>org.zeromq</groupId>
			<artifactId>jzmq</artifactId>
			<version>2.1.0</version>
		</dependency>
	</dependencies>

 

消息发布：

package com.giant.test.jzmq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/**
 * 先启动订阅者：
 * 	java -Djava.library.path=/usr/local/lib -cp $(echo ./*.jar | tr ' ' ':') com.giant.test.jzmq.Subscriber
 * 再启动发送者：
 * 	java -Djava.library.path=/usr/local/lib -cp $(echo ./*.jar | tr ' ' ':') com.giant.test.jzmq.Publisher
* <p>Description: </p>
* @author wankun
* @date 2014年6月9日
* @version 1.0
 */
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();
	}
}

 

消息订阅：

package com.giant.test.jzmq;

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://127.0.0.1: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();
	}
}