package com.coderli.log4j.mdc;


import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;


import org.apache.log4j.MDC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
* 单线程多 classloader模式下，log4jMDC信息传递测试。
*
* @author OneCoder
* @blog http://www.coderli.com
* @date 2014年5月6日 上午10:55:34
*/
public class SingleThreadMultiLoader {


     private static final Logger logger = LoggerFactory
              . getLogger(SingleThreadMultiLoader.class);
     private static String key = "loaderName";
     static ThreadLocal<String> tl = new ThreadLocal<String>();


     public static void main(String[] args) throws MalformedURLException,
              ClassNotFoundException, InstantiationException,
              IllegalAccessException, NoSuchMethodException, SecurityException,
              IllegalArgumentException, InvocationTargetException {
          MDC. put(key, "in main loader" );
           logger.info( "线程名: {}; 日志MDC信息：{}。" , Thread.currentThread().getName(),
                   MDC. get(key));
           tl.set( "huanyige");
          ClassLoader cLoader = Thread.currentThread().getContextClassLoader();
          URL[] urls = new URL[] {
                    new URL("file:\\D:\\acap\\workspace\\shurnim-lab\\lib\\mdc.jar" ),
                    new URL(
                              "file:\\D:\\acap\\workspace\\shurnim-lab\\lib\\log4j-1.2.17.jar" ),
                    new URL(
                              "file:\\D:\\acap\\workspace\\shurnim-lab\\lib\\slf4j-api-1.7.5.jar" ),
                    new URL(
                              "file:\\D:\\acap\\workspace\\shurnim-lab\\lib\\slf4j-log4j12-1.7.5.jar" ) };
          ClassLoader loader = new URLClassLoader(urls, null);
          Thread. currentThread().setContextClassLoader(loader);
          String className = SingleThreadMultiLoader.class.getName();
           Class clz = loader.loadClass(className);
          Method main = clz.getMethod("logMethod");
          main.invoke( null);
           logger.info( "线程名: {}; 日志MDC信息：{}。" , Thread.currentThread().getName(),
                   MDC. get(key));
     }


     public static void logMethod() {
           logger.info( "线程名: {}; 日志MDC信息：{}。" , Thread.currentThread().getName(),
                   MDC. get(key));
          MDC. put(key, "hahahahhaha" );
          System. out.println( tl.get());
     }
}

public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null)
                return (T)e.value;
        }
        return setInitialValue();
    }

package com.coderli.schema;


import java.io.InputStream;
import java.net.URL;


import javax.xml.transform.Source;
import javax.xml.transform.stream.StreamSource;
import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
import javax.xml.validation.Validator;


import org.dom4j.io.OutputFormat;
import org.dom4j.io.XMLWriter;
import org.dom4j.util.XMLErrorHandler;
import org.xml.sax.SAXException;


/**
* @author lihzh
* @date 2014年4月2日 下午2:34:08
*/
public class JobConfigXMLValidateTest {


     public static void main(String args[]) throws SAXException {
           validateXMLByXSD();
     }


     /**
      *
      * @throws SAXException
      * @author lihzh
      * @date 2014年4月2日 下午4:25:39
      */
     public static void validateXMLByXSD() throws SAXException {
          String xmlFileName = "com/coderli/schema/shurnim.xml" ;
          String xsdFileName = "com/coderli/schema/shurnim.xsd" ;
           // 建立schema工厂
          SchemaFactory schemaFactory = SchemaFactory
                    . newInstance("http://www.w3.org/2001/XMLSchema");
           // 建立验证文档文件对象，利用此文件对象所封装的文件进行schema验证
          URL schemaFile = JobConfigXMLValidateTest. class.getClassLoader()
                   .getResource(xsdFileName);
           // 利用schema工厂，接收验证文档文件对象生成Schema对象
          Schema schema = schemaFactory.newSchema(schemaFile);
           // 通过Schema产生针对于此Schema的验证器，利用schenaFile进行验证
          Validator validator = schema.newValidator();
           // 创建默认的XML错误处理器
          XMLErrorHandler errorHandler = new XMLErrorHandler();
          validator.setErrorHandler(errorHandler);
           // 得到验证的数据源
          InputStream xmlStream = JobConfigXMLValidateTest. class.getClassLoader()
                   .getResourceAsStream(xmlFileName);
          Source source = new StreamSource(xmlStream);
           // 开始验证，成功输出success!!!，失败输出fail
           try {
              validator.validate(source);
              XMLWriter writer = new XMLWriter(OutputFormat.createPrettyPrint());
               // 如果错误信息不为空，说明校验失败，打印错误信息
               if (errorHandler.getErrors().hasContent()) {
                   System. out.println( "XML文件通过XSD文件校验失败！" );
                   writer.write(errorHandler.getErrors());
              } else {
                   System. out.println( "Good! XML文件通过XSD文件校验成功！" );
              }
              ;
          } catch (Exception ex) {
              ex.printStackTrace();
          }
     }
}

SAXReader saxReader = new SAXReader();
               Document document = saxReader.read(configFile);
               Element root = document.getRootElement();
               QName qName = new QName("schemaLocation", new Namespace("xsi", "http://www.w3.org/2001/XMLSchema-instance"));
               Attribute xsdAddr = root.attribute(qName);
               System.out.println(xsdAddr.getText());

package com.coderli.zeromq.majordomoprotocol;


import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.Map;


import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Majordomo Protocol协议验证<br>
 * 用于实现多client、多worker实现双向指定目标数据收发 <br>
 * 此为核心Broker模块
 *
 */
public class Broker extends JZMQBase {


     private static class Service {
           // 服务名
           public final String name;
           // 请求信息队列
          Deque<ZMsg> requests;
           // 待用worker队列
          Deque<Worker> waiting; // List of waiting workers


           public Service(String name) {
               this. name = name;
               this. requests = new ArrayDeque<ZMsg>();
               this. waiting = new ArrayDeque<Worker>();
          }
     }


     private static class Worker {
           // worker的唯一标识
           @SuppressWarnings( "unused")
          String identity; // Identity of worker
           // 目标worker地址
          ZFrame address; // Address frame to route to
           // 包含的service名称，如果存在
          Service service;


           public Worker(String identity, ZFrame address) {
               this. address = address;
               this. identity = identity;
          }
     }


     private ZContext ctx;
     private ZMQ.Socket socket;
     private Map<String, Service> services;
     private Map<String, Worker> workers;
     private Deque<Worker> waiting;


     public static void main(String[] args) {
          Broker broker = new Broker();
          broker.bind( BROKER_FRONT_END);
          broker.mediate();
     }


     public Broker() {
           this. services = new HashMap<String, Service>();
           this. workers = new HashMap<String, Worker>();
           this. waiting = new ArrayDeque<Worker>();
           this. ctx = new ZContext();
           this. socket = ctx.createSocket(ZMQ. ROUTER);
     }


     public void mediate() {
           while (!Thread. currentThread().isInterrupted()) {
              ZMQ.Poller items = new ZMQ.Poller(1);
              items.register( socket, ZMQ.Poller. POLLIN);
              items.poll();
               if (items.pollin(0)) {
                   ZMsg msg = ZMsg. recvMsg(socket);
                    if (msg == null) {
                        System. out.println( "接收到的消息为null。" );
                         break; // Interrupted
                   }
                   System. out.println( "I: received message:\n");
                   msg.dump(System. out);
                    // 根据协议栈规则读取数据，此处需要注意broker接受到的协议栈格式
                   ZFrame sender = msg.pop();
                   ZFrame empty = msg.pop();
                   ZFrame header = msg.pop();
                    if (MDP. C_CLIENT.frameEquals(header)) {
                        processClient(sender, msg);
                   } else if (MDP.W_WORKER.frameEquals(header))
                        processWorker(sender, msg);
                    else {
                        System. out.println( "E: invalid message:\n");
                        msg.dump(System. out);
                        msg.destroy();
                   }
                   sender.destroy();
                   empty.destroy();
                   header.destroy();
              }
          }
          destroy();
     }


     private void destroy() {
          Worker[] deleteList = workers.entrySet().toArray( new Worker[0]);
           for (Worker worker : deleteList) {
              deleteWorker(worker, true);
          }
           ctx.destroy();
     }


     /**
      * 处理客户端请求的，用于分发给指定的worker.
      */
     private void processClient(ZFrame sender, ZMsg msg) {
           if (msg.size() < 2) {
              System. out.println( "消息栈不完整，不能发送" );
               return;
          }
          ZFrame serviceFrame = msg.pop();
          msg.wrap(sender);
          dispatch(requireService(serviceFrame), msg);
          serviceFrame.destroy();
     }


     private void processWorker(ZFrame sender, ZMsg msg) {
           if (msg.size() < 1) {
              System. out.println( "回复给客户端的消息不完整，不能发送。" );
          }
          ZFrame command = msg.pop();
           boolean workerReady = workers.containsKey(sender.strhex());
          Worker worker = requireWorker(sender);
           if (MDP. W_READY.frameEquals(command)) {
               if (workerReady) {
                   System. out.println( "删除worker：" + sender.strhex());
                   deleteWorker(worker, true);
              } else {
                   ZFrame serviceFrame = msg.pop();
                   worker. service = requireService(serviceFrame);
                   workerWaiting(worker);
                   serviceFrame.destroy();
              }
          } else if (MDP. W_REPLY.frameEquals(command)) {
               if (workerReady) {
                   System. out.println( "开始给客户端相应" );
                   ZFrame client = msg.unwrap();
                   msg.addFirst(worker. service. name);
                   msg.addFirst(MDP. C_CLIENT.newFrame());
                   msg.wrap(client);
                   msg.send( socket);
                   workerWaiting(worker);
              } else {
                   deleteWorker(worker, true);
              }
          } else {
              System. out.print( "不合法的消息结构" );
              msg.dump(System. out);
          }
          msg.destroy();
     }


     private void deleteWorker(Worker worker, boolean disconnect) {
          System. out.println( "删除worker");
           if (disconnect) {
              sendToWorker(worker, MDP. W_DISCONNECT, null, null);
          }
           if (worker. service != null)
              worker. service. waiting.remove(worker);
           workers.remove(worker);
          worker. address.destroy();
     }


     private Worker requireWorker(ZFrame address) {
           assert (address != null);
          String identity = address.strhex();
          Worker worker = workers.get(identity);
           if (worker == null) {
              worker = new Worker(identity, address.duplicate());
               workers.put(identity, worker);
              System. out.println( "注册了新的worker：" + identity);
          }
           return worker;
     }


     private Service requireService(ZFrame serviceFrame) {
           assert (serviceFrame != null);
          String name = serviceFrame.toString();
          Service service = services.get(name);
           if (service == null) {
              service = new Service(name);
               services.put(name, service);
          }
           return service;
     }


     private void bind(String endpoint) {
           socket.bind(endpoint);
          System. out.println( "Broker版定在端口： " + endpoint);
     }


     public synchronized void workerWaiting(Worker worker) {
           waiting.addLast(worker);
          worker. service. waiting.addLast(worker);
          dispatch(worker. service, null);
     }


     private void dispatch(Service service, ZMsg msg) {
           assert (service != null);
           if (msg != null) {
              service. requests.offerLast(msg);
          }
           while (!service. waiting.isEmpty() &amp;&amp; !service.requests.isEmpty()) {
              msg = service. requests.pop();
              Worker worker = service. waiting.pop();
               waiting.remove(worker);
              sendToWorker(worker, MDP. W_REQUEST, null, msg);
              msg.destroy();
          }
     }


     public void sendToWorker(Worker worker, MDP command, String option,
              ZMsg msgp) {
          ZMsg msg = msgp == null ? new ZMsg() : msgp.duplicate();
           if (option != null)
              msg.addFirst( new ZFrame(option));
          msg.addFirst(command.newFrame());
          msg.addFirst(MDP. W_WORKER.newFrame());
          msg.wrap(worker. address.duplicate());
          System. out.println( "给worker发送命令： [" + command + "]。");
          msg.dump(System. out);
          msg.send( socket);
     }
}

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;


/**
 * ZeroMQ Majordomo Protocol协议验证<br>
 * 用于实现多client、多worker实现双向指定目标数据收发 <br>
 * 此为Client端依赖的API。
 *
 */
public class ClientAPI {


     private String broker;
     private ZContext ctx;
     private ZMQ.Socket client;
     private long timeout = 2500;
     private int retries = 3;


     public long getTimeout() {
           return timeout;
     }


     public void setTimeout( long timeout) {
           this. timeout = timeout;
     }


     public int getRetries() {
           return retries;
     }


     public void setRetries( int retries) {
           this. retries = retries;
     }


     public ClientAPI(String broker) {
           this. broker = broker;
           ctx = new ZContext();
          reconnectToBroker();
     }


     void reconnectToBroker() {
           if ( client != null) {
               ctx.destroySocket( client);
          }
           client = ctx.createSocket(ZMQ. REQ);
           client.connect( broker);
          System. out.println( "连接到Broker：" + broker );
     }


     /**
      * 给broker发送消息
      *
      * @param service
      * @param request
      * @return
      */
     public ZMsg send(String service, ZMsg request) {


          request.push( new ZFrame(service));
          request.push(MDP. C_CLIENT.newFrame());
          System. out.println( "发送消息给worker：" + service);
          request.dump(System. out);
          ZMsg reply = null;


           int retriesLeft = retries;
           while (retriesLeft > 0 &amp;&amp; !Thread.currentThread().isInterrupted()) {
              request.duplicate().send( client);
              ZMQ.Poller items = new ZMQ.Poller(1);
              items.register( client, ZMQ.Poller. POLLIN);
               if (items.poll( timeout) == -1)
                    break; // 超时退出
               if (items.pollin(0)) {
                   ZMsg msg = ZMsg. recvMsg(client);
                   System. out.println( "接收到消息。" );
                   msg.dump(System. out);
                   ZFrame header = msg.pop();
                   header.destroy();
                   ZFrame replyService = msg.pop();
                   replyService.destroy();
                   reply = msg;
                    break;
              } else {
                   items.unregister( client);
                    if (--retriesLeft == 0) {
                        System. out.println( "超过重试次数，错误。退出。" );
                         break;
                   }
                   System. out.println( "没有收到回应，重试。" );
                   reconnectToBroker();
              }
          }
          request.destroy();
           return reply;
     }


     public void destroy() {
           ctx.destroy();
     }
}

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZMsg;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Majordomo Protocol协议验证<br>
 * 用于实现多client、多worker实现双向指定目标数据收发 <br>
 * 此为Client端一号，定向发给1、2号worker
 *
 */
public class ClientOne extends JZMQBase {


     public static void main(String[] args) throws InterruptedException {
          ClientAPI clientSession = new ClientAPI(BROKER_FRONT_END);


           int count;
           for (count = 0; count < 1; count++) {
              ZMsg request = new ZMsg();
              ZMsg reply = null;
               long start = System. nanoTime();
              request.addString(String. valueOf(start));
               if (count % 2 == 1) {
                   reply = clientSession.send( "one", request);
              } else {
                   reply = clientSession.send( "two", request);
              }
               if (reply != null)
                   reply.destroy();
               else
                    break; // Interrupt or failure
              Thread. sleep(1000000L);
          }


          System. out.printf( "%d requests/replies processed\n", count);
          clientSession.destroy();
     }
}

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;


/**
 * ZeroMQ验证 workerAPI封装
 *
 * @author lihzh
 * @date 2014年1月15日 下午2:23:14
 */
public class WorkerAPI {


     private String broker;
     private ZContext ctx;
     private String service;
     private ZMQ.Socket worker;


     private long timeout = 2500;


     private ZFrame replyTo;


     public WorkerAPI(String broker, String service) {
           assert (broker != null);
           assert (service != null);
           this. broker = broker;
           this. service = service;
           ctx = new ZContext();
          reconnectToBroker();
     }


     /**
      * 给Broker发送消息
      *
      * @param command
      * @param option
      * @param msg
      */
     void sendToBroker(MDP command, String option, ZMsg msg) {
          msg = msg != null ? msg.duplicate() : new ZMsg();


           if (option != null)
              msg.addFirst( new ZFrame(option));
          msg.addFirst(command.newFrame());
          msg.addFirst(MDP. W_WORKER.newFrame());
          msg.addFirst( new ZFrame( new byte[0]));
          msg.send( worker);
     }


     void reconnectToBroker() {
           if ( worker != null) {
               ctx.destroySocket( worker);
          }
           worker = ctx.createSocket(ZMQ. DEALER);
           worker.connect( broker);
          sendToBroker(MDP. W_READY, service, null);
     }


     /**
      * 接收数据
      *
      * @param reply
      * @return
      * @author lihzh
      * @date 2014年1月15日 下午2:24:23
      */
     public ZMsg receive(ZMsg reply) {
           if (reply != null) {
              reply.wrap( replyTo);
              sendToBroker(MDP. W_REPLY, null, reply);
              reply.destroy();
          }
           while (!Thread. currentThread().isInterrupted()) {
              ZMQ.Poller items = new ZMQ.Poller(1);
              items.register( worker, ZMQ.Poller. POLLIN);
               if (items.poll( timeout) == -1)
                    break; // Interrupted
               if (items.pollin(0)) {
                   ZMsg msg = ZMsg. recvMsg(worker);
                    if (msg == null)
                         break; // Interrupted
                   System. out.print( "接收到数据：" );
                    long time = System. nanoTime();
                    long endTime = Long
                             . valueOf(new String(msg.getLast().getData()));
                   System. out.println( "消耗时间：" + (time - endTime));
                   msg.dump(System. out);
                   ZFrame empty = msg.pop();
                   empty.destroy();
                   ZFrame header = msg.pop();
                   header.destroy();
                   ZFrame command = msg.pop();
                    if (MDP.W_REQUEST.frameEquals(command)) {
                         replyTo = msg.unwrap();
                        command.destroy();
                         return msg;
                   } else {
                        System. out.println( "不合法的消息结构。" );
                        msg.dump(System. out);
                   }
                   command.destroy();
                   msg.destroy();
              }


          }
           return null;
     }


     public void destroy() {
           ctx.destroy();
     }
}

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZMsg;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Majordomo Protocol协议验证<br>
 * 用于实现多client、多worker实现双向指定目标数据收发 <br>
 * 此为Worker端，定向回复给调用的client
 *
 */
public class WorkerOne extends JZMQBase {


     /**
      * @param args
      */
     public static void main(String[] args) {
          WorkerAPI workerSession = new WorkerAPI(BROKER_FRONT_END, "one" );


          ZMsg reply = null;
           while (!Thread. currentThread().isInterrupted()) {
              ZMsg request = workerSession.receive(reply);
               if (request == null)
                    break;
              reply = request;
          }
          workerSession.destroy();
     }
}

package com.coderli.zeromq.majordomoprotocol;


import org.zeromq.ZMsg;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Majordomo Protocol协议验证<br>
 * 用于实现多client、多worker实现双向指定目标数据收发 <br>
 * 此为Worker端，定向回复给调用的client
 *
 */
public class WorkerTwo extends JZMQBase {


     /**
      * @param args
      */
     public static void main(String[] args) {
          WorkerAPI workerSession = new WorkerAPI(BROKER_FRONT_END, "two" );


          ZMsg reply = null;
           while (!Thread. currentThread().isInterrupted()) {
              ZMsg request = workerSession.receive(reply);
               if (request == null)
                    break;
              reply = request;
          }
          workerSession.destroy();
     }
}

package com.coderli.zeromq.majordomoprotocol;


import java.util.Arrays;


import org.zeromq.ZFrame;


/**
 * ZeroMQ Majordomo Protocol协议验证<br>
 * 用于实现多client、多worker实现双向指定目标数据收发 <br>
 * 此为常量类
 *
 */
public enum MDP {


     C_CLIENT("MDPC01"), W_WORKER("MDPW01"),


     W_READY(1), W_REQUEST(2), W_REPLY(3), W_HEARTBEAT(4), W_DISCONNECT (5);


     private final byte[] data;


     MDP(String value) {
           this. data = value.getBytes();
     }


     MDP(int value) { // watch for ints>255, will be truncated
           byte b = ( byte) (value &amp; 0xFF);
           this. data = new byte[] { b };
     }


     public ZFrame newFrame() {
           return new ZFrame( data);
     }


     public boolean frameEquals(ZFrame frame) {
           return Arrays. equals(data, frame.getData());
     }
}

/**
 * @author lihzh
 * @date 2014年1月14日 上午9:32:01
 */
public abstract class JZMQBase {


     protected static String LOCAL_ADDRESS = "tcp://127.0.0.1:1234";
     protected static String LOCAL_ADDRESS_PUSHER = "tcp://127.0.0.1:2345";
     protected static String LOCAL_ADDRESS_ROUTER = "tcp://127.0.0.1:3456";
     protected static String LOCAL_ADDRESS_DECLARER = "tcp://127.0.0.1:4567";


     protected static String BROKER_FRONT_END = "tcp://127.0.0.1:4000";
     protected static String BROKER_BACK_END = "tcp://127.0.0.1:4001";
}

package com.coderli.zeromq.requestreplay;


import org.zeromq.ZMQ;
import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ 请求响应模式验证代码 <br>
 * 此为服务端
 *
 * @author OneCoder
 * @date 2014年1月13日 下午11:28:47
 * @website http://www.coderli.com
 */
public class ReplayServer extends JZMQBase {


     public static void main(String[] args) {
           // 参数代表使用多少线程，大多数情况下，1个线程已经足够。
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为响应模式
          ZMQ.Socket socket = context.socket(ZMQ. REP);
          socket.bind( LOCAL_ADDRESS); // 绑定服务地址及端口
           for (;;) {
              System. out.println( "Server start.");
              socket.recv();
              String str = "Ok, I'm server";
              socket.send(str);
          }
     }
}

package com.coderli.zeromq.requestreplay;


import org.zeromq.ZMQ;
import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ 请求响应模式验证代码 <br>
 * 此为客户端
 *
 * @author OneCoder
 * @date 2014年1月11日 上午10:34:18
 * @website http://www.coderli.com
 */
public class RequestClient extends JZMQBase {


     /**
      * @param args
      * @author OneCoder
      * @date 2014年1月11日 上午10:34:18
      */
     public static void main(String[] args) {


          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为请求模式
          ZMQ.Socket socket = context.socket(ZMQ. REQ);
           // 创建链接
          socket.connect( LOCAL_ADDRESS);
           int count = 1;
           for (;;) {
               try {
                    long time = System. nanoTime();
                   socket.send( "Hello, currentTime: " + count);
                    byte[] recs = socket.recv();
                    long end = System. nanoTime();
                   System. out.println( new String(recs) + " Cost time: "
                             + (end - time));
                   count++;
                   Thread. sleep(1000);
              } catch (Exception e) {
                   e.printStackTrace();
              }
          }
     }
}

package com.coderli.zeromq.pubsub;


import java.util.Random;


import org.zeromq.ZMQ;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ 发布订阅模式Java验证代码 <br>
 * 此为发布者
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class Publisher extends JZMQBase {


     public static void main(String[] args) throws InterruptedException {
           // 参数代表使用多少线程，大多数情况下，1个线程已经足够。
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为发布者
          ZMQ.Socket socket = context.socket(ZMQ. PUB);
          socket.bind( LOCAL_ADDRESS); // 绑定服务地址及端口
           for (;;) {
               int i = ( int) ( new Random().nextDouble() * 2 + 1);
              String s = String. valueOf(i);
               long time = System. nanoTime();
              socket.send(s + String. valueOf(time));
              System. out.println( "发布了新消息，时间：" + time + " 类型：" + s);
              Thread. sleep(2000);
          }
     }
}

package com.coderli.zeromq.pubsub;


import org.zeromq.ZMQ;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ 发布订阅模式Java验证代码 <br>
 * 此为订阅者1号
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class SubscriberOne extends JZMQBase {


     public static void main(String[] args) throws InterruptedException {
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为请求模式
          ZMQ.Socket socket = context.socket(ZMQ. SUB);
           // 创建订阅者，必须要过主题过滤器
           byte[] filter = "1".getBytes();
          socket.subscribe(filter);
          socket.connect( LOCAL_ADDRESS);
           for (;;) {
               byte[] recs = socket.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);
          }
     }
}

package com.coderli.zeromq.pubsub;


import org.zeromq.ZMQ;


import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ 发布订阅模式Java验证代码 <br>
 * 此为订阅者2号
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class SubscriberTwo extends JZMQBase{


     public static void main(String[] args) throws InterruptedException {
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为请求模式
          ZMQ.Socket socket = context.socket(ZMQ. SUB);
           // 创建订阅者，必须要过主题过滤器
           byte[] filter = "2".getBytes();
          socket.subscribe(filter);
          socket.connect( LOCAL_ADDRESS);
           for (;;) {
               byte[] recs = socket.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);
          }
     }
}

package com.coderli.zeromq.pipeline;

import org.zeromq.ZMQ;
import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Pipeline模式Java验证代码 <br>
 * 此为主Pusher
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class MainPusher extends JZMQBase {


     public static void main(String[] args) throws InterruptedException {
           // 参数代表使用多少线程，大多数情况下，1个线程已经足够。
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为Pusher
          ZMQ.Socket socket = context.socket(ZMQ. PUSH);
          socket.bind( LOCAL_ADDRESS); // 绑定服务地址及端口
           for (;;) {
               long time = System. nanoTime();
              socket.send(String. valueOf(time));
              System. out.println( "发布了新消息，时间：" + time);
              Thread. sleep(2000);
          }
     }


}

package com.coderli.zeromq.pipeline;

import org.zeromq.ZMQ;
import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Pipeline模式Java验证代码 <br>
 * 此为中转worker
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class WorkerOne extends JZMQBase {


     public static void main(String[] args) {
           // 指定模式为pull模式
          ZMQ.Socket receiver = ZMQ.context(1).socket(ZMQ.PULL);
          receiver.connect( LOCAL_ADDRESS);
           // 指定模式为push模式
          ZMQ.Socket sender = ZMQ.context(1).socket(ZMQ.PUSH);
          sender.connect( LOCAL_ADDRESS_PUSHER);
           for (;;) {
               byte[] recs = receiver.recv();
               long receiveTime = System. nanoTime();
              String oriMsg = new String(recs);
               long pubTime = Long. valueOf(oriMsg);
               long costTime = receiveTime - pubTime;
              System. out.println( "Receive: " + oriMsg + " Cost time: " + costTime);
              sender.send( "1" + oriMsg);
              System. out.println( "Send to sinker.");
          }
     }
}

package com.coderli.zeromq.pipeline;

import org.zeromq.ZMQ;
import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Pipeline模式Java验证代码 <br>
 * 此为中转worker
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class WorkerTwo extends JZMQBase {


     public static void main(String[] args) {
           // 指定模式为pull模式
          ZMQ.Socket receiver = ZMQ.context(1).socket(ZMQ.PULL);
          receiver. connect(LOCAL_ADDRESS);
           // 指定模式为push模式
          ZMQ.Socket sender = ZMQ.context(1).socket(ZMQ.PUSH);
          sender. connect(LOCAL_ADDRESS_PUSHER);
           for (;;) {
               byte[] recs = receiver.recv();
               long receiveTime = System. nanoTime();
              String oriMsg = new String(recs);
               long pubTime = Long. valueOf(oriMsg);
               long costTime = receiveTime - pubTime;
              System. out
                        .println( "Receive: " + oriMsg + " Cost time: " + costTime);
              sender.send( "2" + oriMsg);
              System. out.println( "Send to sinker.");
          }
     }
}

package com.coderli.zeromq.pipeline;

import org.zeromq.ZMQ;
import com.coderli.zeromq.JZMQBase;


/**
 * ZeroMQ Pipeline模式Java验证代码 <br>
 * 此为最终sinker
 *
 * @author OneCoder
 * @date 2014年1月14日 上午11:16:03
 * @blog http://www.coderli.com
 */
public class Sinker extends JZMQBase {


     public static void main(String[] args) {
          ZMQ.Context context = ZMQ. context(1);
           // 指定模式为pull模式
          ZMQ.Socket receiver = context.socket(ZMQ. PULL);
          receiver. bind(LOCAL_ADDRESS_PUSHER);
           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);
          }
     }
}