java zmq消息队列

喜欢ヅ旅行 2022-07-15 03:56 364阅读 0赞

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();    
        \}