java zmq订阅_java zmq消息队列
ZMQ(以下ZeroMQ简称ZMQ)是一个消息队列 ，可以在进程内、进程间、TCP、多播中，以消息为单位传输数据，而不是socket的字节流；同时它像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高.
消息系统的核心作用就是三点：解耦，异步和并行
集群对外提供服务的过程中要保证信息从一个节点，将信息无误的分发到各个服务器节点上，并保证信息正确性和一致性。
1.串行方式的最大问题是，随着后端流程越来越多，每步流程都需要额外的耗费很多时间，从而会导致用户更长的等待延迟。
本人测试使用的是Win7 64位的电脑，jzmq.dll，libzmq.dll放在自己JDK的bin目录下，zmq.jar放在工程lib下；
通信模式：1.发送端
/**
* @throws InterruptedException
* @描述:1.非阻塞发送
* @date:2016年11月1日
*/
public static void zmq_pub_NOBLOCK() throws InterruptedException{
//开启的io线程个数,默认1
Context context = ZMQ.context(1);
//选择发布模式
Socket publisher = context.socket(ZMQ.PUB);
//绑定IP端口
publisher.bind("tcp://*:5561");
//zmq发送速度太快，在订阅者尚未与发布者建立联系时，已经开始了数据发布
Thread.sleep(1000); //需要休息一下
int update_nbr;
for (update_nbr = 20; update_nbr < 40; update_nbr++) {
String a="{\"magicNum\":\"CHINSOFT\",\"varName\":\"ZJ_YD_1\",\"varType\":\"5\",\"varValue\":"+update_nbr+",\"varQuality\":\"1111\",\"varTime\":"+System.currentTimeMillis()/1000+"}";
publisher.send(a.getBytes(), ZMQ.NOBLOCK); //发送，非阻塞模式(ZMQ.SNDMORE：指出当前正在发送的消息是个多帧消息，并且接下来还会发送更多的消息;)
System.out.println(update_nbr);
Thread.sleep(1000);
}
//关闭本进程的socket id，但链接还是开着的，用这个socket id的其它进程还能用这个链接，能读或写这个socket id
publisher.close();
//函数会进入阻塞状态，直到满足下列条件,所有基于context创建的scoekt都已经被zmq_close()函数关闭
context.term();
}
//订阅客户端
public static void main(String[] args) {
//开启的io线程个数,默认1
Context context = ZMQ.context(1);
// PUB/SUB模式: 发布/订阅模式
Socket subscriber = context.socket(ZMQ.SUB);
subscriber.connect("tcp://localhost:5561");
//设置订阅条件"setsockopt"
subscriber.subscribe("".getBytes());
int update_nbr = 0;
while (true) {
byte[] stringValue = subscriber.recv(0);
String string = new String(stringValue);
update_nbr++;
//{"magicNum":"CHINSOFT","varName":"ZJ_YD_1","varType":"5","varValue":20,"varQuality":"1111","varTime":1477963955}
System.out.println("Received " + update_nbr + " updates. :"+ string);
}
}
/**
* @描述:消息队列 -多针发送,
* @date:2016年11月1日
* @throws InterruptedException
*/
public static void zmq_pub_multiSend() throws InterruptedException{
Context context = ZMQ.context(1);
Socket publisher = context.socket(ZMQ.PUB);
publisher.bind("tcp://*:5561");
int i = 0;
while (i<10) {
i++;
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);
}
}
public static void sub_1(){
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)));
}
}
public static void sub_2(){
Context context = ZMQ.context(1);
Socket subscribe = context.socket(ZMQ.SUB);
subscribe.connect("tcp://127.0.0.1:5561");
//subscribe.subscribe("topic".getBytes());
subscribe.subscribe("A".getBytes());
while (true) {
System.out.println(new String(subscribe.recv(0)));
System.out.println(new String(subscribe.recv(0)));
}
}
public static void main(String[] args) {
for (int j = 0; j < 100; j++) {
new Thread(new Runnable(){
public void run() {
// TODO Auto-generated method stub
ZMQ.Context context = ZMQ.context(1); //创建1个I/O线程的上下文
ZMQ.Socket subscriber = context.socket(ZMQ.SUB); //创建一个sub类型，也就是subscriber类型的socket
subscriber.connect("tcp://127.0.0.1:5561"); //与在5555端口监听的publisher建立连接
subscriber.subscribe("fjs".getBytes()); //订阅fjs这个channel
for (int i = 0; i < 100; i++) {
byte[] message = subscriber.recv(); //接收publisher发送过来的消息
System.out.println("receive : " + new String(message));
}
subscriber.close();
context.term();
}
}).start();
}