ZeroMQ(java)之Router/Dealer模式

ZeroMQ(java)之Router/Dealer模式

2014-02-28 23:12
9504人阅读 评论(1)
收藏
举报


分类：
网络编程（82）


在开始之前先把guid里面提到的几个ZeroMQ的特性列一下吧：

（1）ZeroMQ有自己的I/O线程来异步的处理I/O，而且后台采用了无锁的数据结构

（2）在ZeroMQ中，所有的组件都可以动态的加入和移除，而且可以启动组件以任何的顺利，例如我们可以先启动request，再启动response，依然可以工作，而且还会自动的重连接。

（3）如果有需要的话，会自动的将message进行排队，当然这都是有一套的模式的，一般情况下会尽量早的将数据发送到receiver。

（4）当缓冲的message队列满了以后，ZeroMQ有自己的行为，有的组件会阻塞，有的则会将message抛弃。

（5）底层的通信可以采用各种各样的都行，例如TCP，IPC啥的。

（6）它会自动的处理那些比较慢而且阻塞的reader

（7）支持message的路由

（8）ZeroMQ确保全部的数据被receiver接收到，例如发送10K，那么也接受到10K

（9）它发送的数据格式是二进制，所以对发送的内容无要求

（10）ZeroMQ会自动处理网络错误，而且会自动尝试恢复

（11）节能。。。（我擦，居然还有这个）

好了，先来看一下poller这个东西吧，蛮有意思的，类似与epoll或者Java里面的selector，

在前面的例子中我们都只是创建一个socket，那如果我们要创建两个socket在同一个线程中该怎么处理呢，那么这个时候就可以用到poller这东西了。。。可以将已经建立好的socket注册到poller上面去，并注册相应的事件。。

这里就用push/pull来举例子吧，就直接来看pull端的代码吧：

[java] view
plaincopy





package poller;  

  

import org.zeromq.ZMQ;  

  

public class Pull {  

    public static void main(String args[]) {  

        ZMQ.Context context = ZMQ.context(1);  

          

        ZMQ.Socket pull1 = context.socket(ZMQ.PULL);  //创建一个pull  

        pull1.connect("tcp://127.0.0.1:5555");    //建立于push的连接  

        ZMQ.Socket pull2 = context.socket(ZMQ.PULL);  

        pull2.connect("tcp://127.0.0.1:5555");  

          

        ZMQ.Poller poller = new ZMQ.Poller(2);  //创建一个大小为2的poller  

        poller.register(pull1, ZMQ.Poller.POLLIN);  //分别将上述的pull注册到poller上，注册的事件是读  

        poller.register(pull2, ZMQ.Poller.POLLIN);  

        int i = 0;  

        while (!Thread.currentThread().isInterrupted()) {  

            poller.poll();  

            if (poller.pollin(0)) {  

                while (null != pull1.recv(ZMQ.NOBLOCK)) {  //这里采用了非阻塞，确保一次性将队列中的数据读取完  

                    i++;  

                }  

                  

            }  

            if (poller.pollin(1)) {  

                while (null != pull2.recv(ZMQ.NOBLOCK)) {  

                    i++;  

                }  

                  

                  

            }  

            if (i % 10000000 == 0) {  

                System.out.println(i);  

            }  

        }  

        pull1.close();  

        pull2.close();  

        context.term();  

          

    }  

}  

这里还算简单吧，同时创建了两个pull，都将他们注册到了poller上面去。。。其实这个样子很像是selector或者epoll啥的。。。

好啦，接下来进入正题：

request/response算是一种非常常用的模式了，但是前面的例子中，我们的response端都只能在单线程中运行，因为必须要recv与send配对使用，那么就很大程度上限制了response的伸缩性，如果有大量的request来提交很多request请求的话，那么性能将会受到极大的限制，当然这种情况下我们可以采用如下的方式来解决：



这里让request同时连接到多个response，这里就可以将request请求分散到多个response，这样当有多个request的时候的性能要求。。。但是有一个问题，如果我们又10个request端，他们每一个都不断的提交request请求，这个时候我们的reponse可能就会很忙，那么在这种结构下无法动态的添加response，依然限制了整个系统的伸缩性。。。

那么最终的解决方案就来了：



这里可以看到，在request端与response端之间加了一个中间层，可以将其看成一个路由器，它将request端的请求路由到response端，如果性能不够的话，可以再建立新的response将其连接到中间层就可以了，就方便的解决系统的伸缩性问题了。。。

好了，这里直接就上中间层与response端的代码吧：

[java] view
plaincopy





package multireqrep;  

  

import org.zeromq.ZMQ;  

  

public class Response {  

    public static void main(String args[]) {  

        final ZMQ.Context context = ZMQ.context(1);  

        ZMQ.Socket router = context.socket(ZMQ.ROUTER);  

        ZMQ.Socket dealer = context.socket(ZMQ.DEALER);  

          

        router.bind("ipc://fjs1");  

        dealer.bind("ipc://fjs2");  

          

        for (int i = 0; i < 20; i++) {  

            new Thread(new Runnable(){  

  

                public void run() {  

                    // TODO Auto-generated method stub  

      

                    ZMQ.Socket response = context.socket(ZMQ.REP);  

                    response.connect("ipc://fjs2");  

                    while (!Thread.currentThread().isInterrupted()) {  

                        response.recv();  

                        response.send("hello".getBytes());  

                        try {  

                            Thread.currentThread().sleep(1);  

                        } catch (InterruptedException e) {  

                            // TODO Auto-generated catch block  

                            e.printStackTrace();  

                        }  

                    }  

                    response.close();  

                }  

                  

            }).start();  

        }  

        ZMQ.proxy(router, dealer, null);  

        router.close();  

        dealer.close();  

        context.term();  

    }  

}  

好吧，代码还算蛮简单的，直接用了ZeroMQ定义的router和dealer组件，以及内置的proxy方法就好了。。。

嗯，再来赞叹一次，ZeroMQ确实好用。。。