jeroMq示例之 [req-broker-rep]

package guide;

import java.nio.charset.Charset;
import java.text.SimpleDateFormat;
import java.util.Calendar;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

/**
* Simple request-reply broker
*
*/
public class rrbroker{

    public static void main (String[] args) {
        //  Prepare our context and sockets
        Context context = ZMQ.context(1);

        Socket frontend = context.socket(ZMQ.ROUTER);
        Socket backend  = context.socket(ZMQ.DEALER);
        frontend.bind("tcp://*:5559");
        backend.bind("tcp://*:5560");

        System.out.println("launch and connect broker.");

        //  Initialize poll set
        Poller items = new Poller (2);
        items.register(frontend, Poller.POLLIN);
        items.register(backend, Poller.POLLIN);

        boolean more = false;
        byte[] message;

        //  Switch messages between sockets
        while (!Thread.currentThread().isInterrupted()) {            
            //  poll and memorize multipart detection
            items.poll();

            if (items.pollin(0)) {
                while (true) {
                    // receive message
                    message = frontend.recv(0);
                    more = frontend.hasReceiveMore();

                    // Broker it
                    backend.send(message, more ? ZMQ.SNDMORE : 0);
                    
                    String str_msg="";
                    if (message != null) {
                        str_msg= new String(message, Charset.forName("UTF-8"));
                    }
                    System.out.println(GetCurrtime()+": routed a Req. msg["+str_msg+"] more ["+more+"]");
                    if(!more){
                        break;
                    }
                }
            }
            if (items.pollin(1)) {
                while (true) {
                    // receive message
                    message = backend.recv(0);
                    more = backend.hasReceiveMore();
                    // Broker it
                    frontend.send(message,  more ? ZMQ.SNDMORE : 0);
                    System.out.println(GetCurrtime()+": got a Rep.");
                    if(!more){
                        break;
                    }
                }
            }
        }
        //  We never get here but clean up anyhow
        frontend.close();
        backend.close();
        context.term();
    }
    
    public static String GetCurrtime(){
   Calendar cal = Calendar.getInstance();
cal.getTime();
SimpleDateFormat sdf = new SimpleDateFormat("mm:ss.SS");
return sdf.format(cal.getTime());
    }
}

//以下是client程序：

package guide;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/**
* Hello World client
* Connects REQ socket to tcp://localhost:5559
* Sends "Hello" to server, expects "World" back
*/
public class rrclient{

    public static void main (String[] args) {
        Context context = ZMQ.context(1);

        //  Socket to talk to server
        Socket requester = context.socket(ZMQ.REQ);
        requester.connect("tcp://localhost:5559");
        
        System.out.println("launch and connect client.");

        for (int request_nbr = 0; request_nbr < 20; request_nbr++) {
            requester.send("Hello", 0);
            String reply = requester.recvStr (0);
            System.out.println("Received reply " + request_nbr + " [" + reply + "]");
        }
        
        //  We never get here but clean up anyhow
        requester.close();
        context.term();
    }
}

//worker程序：
package guide;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

//  Hello World worker
//  Connects REP socket to tcp://*:5560
//  Expects "Hello" from client, replies with "World"
public class rrworker
{
    public static void main (String[] args) throws Exception {
        Context context = ZMQ.context (1);

        //  Socket to talk to server
        Socket responder = context.socket (ZMQ.REP);
        responder.connect ("tcp://localhost:5560");

        while (!Thread.currentThread ().isInterrupted ()) {
            //  Wait for next request from client
            String string = responder.recvStr (0);
            System.out.printf ("Received request: [%s]\n", string);

            //  Do some ‘work‘
            Thread.sleep (1000);

            //  Send reply back to client
            responder.send ("World");
        }

        //  We never get here but clean up anyhow
        responder.close();
        context.term();
    }
}