import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
protected void initZmq(int clientRequestPort, int workerResponsePort)
{
	Context zmqContext = ZMQ.context(1);
	clientSocket = zmqContext.socket(ZMQ.ROUTER);
	workerSocket = zmqContext.socket(ZMQ.ROUTER);

	clientSocket.setLinger(0);
	workerSocket.setLinger(0);

	clientSocket.setReceiveTimeOut(100);
	workerSocket.setReceiveTimeOut(100);

	clientSocket.bind("tcp://*:" + clientRequestPort);
	workerSocket.bind("tcp://*:" + workerResponsePort);

	poll = new Poller(2);
	poll.register(clientSocket, Poller.POLLIN);
	poll.register(workerSocket, Poller.POLLIN);

	logger.info("Zmq proxy binds to client port [" + clientRequestPort + "] and worker port ["
			+ workerResponsePort + "]");
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
/**
 * @param port
 * @param timeout
 */
public ZmqResponder(int port, Integer timeout)
{
	this.timeout = timeout == null ? -1 : timeout;
	context = ZMQ.context(1);
	// Socket to talk to server
	responder = context.socket(ZMQ.REP);
	responder.setLinger(0);
	String url = String.format("tcp://localhost:%d", port);
	responder.connect(url);
	poller = new Poller(1);
	poller.register(responder, Poller.POLLIN);
	if (timeout != null)
	{
		responder.setReceiveTimeOut(timeout);
		responder.setSendTimeOut(timeout);
	}
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
private void rebuild() {
    pollact = null;

    pollSize = pollers.size();
    pollset = new Poller(pollSize);

    pollact = new SPoller[pollSize];

    int itemNbr = 0;
    for (SPoller poller : pollers) {
        pollset.register(poller.item);
        pollact[itemNbr] = poller;
        itemNbr++;
    }
    dirty = false;
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
@Test
public void testPollerUnregister() {
    Context context = ZMQ.context(1);
    Socket socketOne = context.socket(ZMQ.SUB);
    Socket socketTwo = context.socket(ZMQ.REP);
    Poller poller = new ZMQ.Poller(2);
    poller.register(socketOne, ZMQ.Poller.POLLIN);
    poller.register(socketTwo, ZMQ.Poller.POLLIN);

    socketOne.setLinger(0);
    socketOne.close();
    socketTwo.setLinger(0);
    socketTwo.close();

    poller.unregister(socketOne);
    poller.unregister(socketTwo);
    
    context.term();
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
private void rebuild()
{
    pollact = null;

    pollSize = pollers.size();
    pollset = new Poller(pollSize);

    pollact = new SPoller[pollSize];

    int itemNbr = 0;
    for (SPoller poller : pollers) {
        pollset.register(poller.item);
        pollact[itemNbr] = poller;
        itemNbr++;
    }
    dirty = false;
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
public void run() {
    ZContext ctx = new ZContext();
    Socket client = ctx.createSocket(ZMQ.DEALER);

    //  Set random identity to make tracing easier
    String identity = String.format("%04X-%04X", rand.nextInt(), rand.nextInt());
    client.setIdentity(identity.getBytes(ZMQ.CHARSET));
    client.connect("tcp://localhost:5570");

    PollItem[] items = new PollItem[] { new PollItem(client, Poller.POLLIN) };

    int requestNbr = 0;
    while (!Thread.currentThread().isInterrupted()) {
        //  Tick once per second, pulling in arriving messages
        for (int centitick = 0; centitick < 100; centitick++) {
            ZMQ.poll(items, 10);
            if (items[0].isReadable()) {
                ZMsg msg = ZMsg.recvMsg(client);
                msg.getLast().print(identity);
                msg.destroy();
            }
        }
        client.send(String.format("request #%d", ++requestNbr), 0);
    }
    ctx.destroy();
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
private Socket getSocket() {
  if (socket == null) {
    socket = getZContext().createSocket(ZMQ.DEALER);
    socket.connect(host + ":" + port);
    getPoller().register(socket, Poller.POLLIN);
  }
  return socket;
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
public ServiceControlListener(ServiceController controller) throws ServiceException
{
	this.controller = controller;
	// init zmq
	zmqContext = ZMQ.context(1);
	router = initRouter();
	subscriber = initSubscriber();
	poller = new Poller(2);
	poller.register(router, Poller.POLLIN);
	poller.register(subscriber, Poller.POLLIN);
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
protected void initZmq(byte[] id, String uri, Integer timeout)
{
	this.timeout = timeout == null ? -1 : timeout;
	context = ZMQ.context(1);
	dealer = context.socket(ZMQ.DEALER);
	if (id != null)
	{
		dealer.setIdentity(id);
	}
	dealer.setLinger(0);
	dealer.connect(uri);
	poller = new Poller(1);
	poller.register(dealer, Poller.POLLIN);
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
/**
 * @param port
 * @param timeout
 */
public ZmqRouter(int port, Integer timeout)
{
	this.timeout = timeout == null ? -1 : timeout;
	String connect_str = String.format("tcp://*:%d", port);
	context = ZMQ.context(1);
	router = context.socket(ZMQ.ROUTER);
	router.setLinger(0);
	router.bind(connect_str);
	poller = new Poller(1);
	poller.register(router, Poller.POLLIN);
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
/**
 * @param id
 * @param uri
 * @param timeout
 */
protected void initZmq(byte[] id, String uri, Integer timeout)
{
	this.timeout = timeout == null ? -1 : timeout;
	context = ZMQ.context(1);
	dealer = context.socket(ZMQ.DEALER);
	if (id != null)
	{
		dealer.setIdentity(id);
	}
	dealer.setLinger(0);
	dealer.connect(uri);
	poller = new Poller(1);
	poller.register(dealer, Poller.POLLIN);
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
/**
 * @param host
 * @param port
 * @param topic
 * @param timeout
 */
public ZmqSubscriber(String host, int port, String topic, Integer timeout)
{
	this.timeout = timeout == null ? -1 : timeout;
	String connect_str = String.format("tcp://%s:%d", host, port);
	context = ZMQ.context(1);
	sub = context.socket(ZMQ.SUB);
	sub.subscribe(topic.getBytes());
	sub.connect(connect_str);
	poller = new Poller(1);
	poller.register(sub, Poller.POLLIN);

}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
/**
 * @param host
 * @param port
 * @param timeout
 */
public ZmqPuller(String host, int port, Integer timeout)
{
	this.timeout = timeout == null ? -1 : timeout;
	String connect_str = String.format("tcp://%s:%d", host, port);
	context = ZMQ.context(1);
	receiver = context.socket(ZMQ.PULL);
	receiver.connect(connect_str);
	poller = new Poller(1);
	poller.register(receiver, Poller.POLLIN);
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
protected Object handleRequestMessage(final Message<?> requestMessage) {
	if (!running) {
		return null;
	}
	
	Future<Object> response = executorService.submit(new Callable<Object>() {
		public Object call() throws Exception {
			byte[] requestData = requestConverter.convert(requestMessage.getPayload());
			int retriesLeft = retryCount;
			while (!Thread.currentThread().isInterrupted()) {
				socket.send(requestData);
				PollItem items[] = { new PollItem(socket, Poller.POLLIN) };
				int rc = ZMQ.poll(items, socketReceiveTimeout);
				if (rc == -1) {
					break;
				}
				if (items[0].isReadable()) {
					byte[] reply = socket.recv();
					return replyConverter.convert(reply);
				} else if (--retriesLeft == 0) {
					break;
				} else {
					ZmqLazyPirateGateway.this.connect();
				}
			}
			ZmqLazyPirateGateway.this.connect();
			return null;
		}
	});
			
	try {
		return response.get();
	} catch (Throwable t) {
		throw new MessageHandlingException(requestMessage, t);
	}
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
public static void main (String[] args)
{
    ZContext ctx = new ZContext ();

    //  Use the Zyre UDP class to make sure we listen on the same
    //  network interface as our peers
    ZreUdp udp = new ZreUdp (ZreInterface.PING_PORT_NUMBER);
    String host = udp.host ();
    Socket collector = ctx.createSocket (ZMQ.SUB);

    //  Bind to an ephemeral port
    int port = collector.bindToRandomPort (String.format ("tcp://%s", host));

    //  Announce this to all peers we connect to
    ZreInterface inf = new ZreInterface ();
    inf.setHeader ("X-ZRELOG", "tcp://%s:%d", host, port);

    //  Get all log messages (don't filter)
    collector.subscribe ("".getBytes ());

    Poller items = ctx.getContext ().poller ();
    
    items.register (collector, Poller.POLLIN);
    items.register (inf.handle (), Poller.POLLIN);
    
    while (!Thread.currentThread ().isInterrupted ()) {
        if (items.poll (1000) == -1)
            break;              //  Interrupted
        //  Handle input on collector
        if (items.pollin (0))
            printLogMsg (collector);

        //  Handle event from interface (ignore it)
        if (items.pollin (1)) {
            ZMsg msg = inf.recv ();
            if (msg == null)
                break;              //  Interrupted
            msg.destroy ();
        }
    }
    inf.destroy ();
    udp.destroy ();
    ctx.destroy ();

}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
public static void main(String[] argv)
{
    //  First argument is this broker's name
    //  Other arguments are our peers' names
    //
    if (argv.length < 1) {
        System.out.println("syntax: peering1 me {you}\n");
        System.exit(-1);
    }
    String self = argv[0];
    System.out.println(String.format("I: preparing broker at %s\n", self));
    Random rand = new Random(System.nanoTime());

    ZContext ctx = new ZContext();

    //  Bind state backend to endpoint
    Socket statebe = ctx.createSocket(ZMQ.PUB);
    statebe.bind(String.format("ipc://%s-state.ipc", self));

    //  Connect statefe to all peers
    Socket statefe = ctx.createSocket(ZMQ.SUB);
    statefe.subscribe(ZMQ.SUBSCRIPTION_ALL);
    int argn;
    for (argn = 1; argn < argv.length; argn++) {
        String peer = argv[argn];
        System.out.printf("I: connecting to state backend at '%s'\n", peer);
        statefe.connect(String.format("ipc://%s-state.ipc", peer));
    }
    //  The main loop sends out status messages to peers, and collects
    //  status messages back from peers. The zmq_poll timeout defines
    //  our own heartbeat:

    while (true) {
        //  Poll for activity, or 1 second timeout
        PollItem items[] = {new PollItem(statefe, Poller.POLLIN)};
        int rc = ZMQ.poll(items, 1000);
        if (rc == -1)
            break;              //  Interrupted

        //  Handle incoming status messages
        if (items[0].isReadable()) {
            String peer_name = new String(statefe.recv(0), ZMQ.CHARSET);
            String available = new String(statefe.recv(0), ZMQ.CHARSET);
            System.out.printf("%s - %s workers free\n", peer_name, available);
        } else {
            //  Send random values for worker availability
            statebe.send(self, ZMQ.SNDMORE);
            statebe.send(String.format("%d", rand.nextInt(10)), 0);
        }
    }
    ctx.destroy();
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
private Poller getPoller() {
  if (poller == null) {
    poller = getZContext().createPoller(1);
  }
  return poller;
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
private AldaResponse sendRequest(ZMsg request, int timeout, int retries)
  throws NoResponseException {
  // When non-null, used to help ensure that we don't use a response from the
  // server that was for a different request.
  String jobId = options == null ? null : options.jobId;

  while (retries >= 0 && !Thread.currentThread().isInterrupted()) {
    Poller poller = getPoller();
    Socket client = rebuildSocket();

    // false means don't destroy the message after sending
    request.send(client, false);

    int rc = poller.poll(timeout);
    if (rc == -1) {
      request.destroy();
      throw new NoResponseException("Connection interrupted.");
    }

    if (poller.pollin(0)) {
      ZMsg zmsg = ZMsg.recvMsg(client);
      if (debug) zmsg.dump();

      byte[] address = zmsg.unwrap().getData(); // discard envelope
      String responseJson = zmsg.popString();

      AldaResponse response = AldaResponse.fromJson(responseJson);

      // If there is a jobId option, we will ignore any response from the
      // server that doesn't have the same jobId, and try again. This will not
      // count against our remaining retries, as the server did respond.
      if (jobId != null &&
          response.jobId != null &&
          !jobId.equals(response.jobId))
        continue;

      if (!response.noWorker)
        response.workerAddress = zmsg.pop().getData();

      request.destroy();
      return response;
    }

    // Didn't get a response within the allowed timeout. Try again, unless
    // we're out of retries.
    retries--;
  }

  request.destroy();
  String errorMsg = "Alda server is down. To start the server, run `alda up`.";
  throw new NoResponseException(errorMsg);
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
/**
 * Broadcast a command to a remote server.
 *
 * <p>Example: Send a signature command to the alternate cosigner server. It will respond with
 * signed data.
 *
 * @param command Command to broadcast.
 * @param server  Server to attempt to send the command to.
 * @return Reply from the server.
 */
public static String broadcastCommand(BaseCommand command, Server server) {
  // Update the comm time if we've actually been talking to this server.
  if (ClusterInfo.getInstance().getServers().contains(server)) {
    ClusterInfo.getInstance().getServers().forEach(trackedServer -> {
      if (trackedServer.equals(server)) {
        server.setLastCommunication(trackedServer.getLastCommunication());
      }
    });
  }

  // If we haven't heard from the server in more then 2 minutes, consider it offline.
  if ((System.currentTimeMillis() - server.getLastCommunication()) > 2 * 60 * 1000) {
    if (command.getClass() == ClusterCommand.class
        && ((ClusterCommand) command).getCommandType() == ClusterCommandType.HEARTBEAT) {
      LOGGER.debug("Server is too old, sending heartbeat");
    } else {
      LOGGER.debug("Server is too old, removing server");
      ClusterInfo.getInstance().getServers().remove(server);
      return "";
    }
  }

  // If it's not a ClusterCommand, encrypt it. ClusterCommands only identify servers, and
  // they're checked for signatures.
  if (command.getClass() != ClusterCommand.class) {
    command =
        new EncryptedCommand(ClusterInfo.getInstance().getThisServer(), ServerKey.getMykey(),
            server, command.toJson());
  }

  String commandString = command.toJson();

  Context context = ZMQ.context(1);
  Socket requester = context.socket(ZMQ.REQ);
  requester.connect("tcp://" + server.getServerLocation() + ":" + server.getServerRpcPort());

  requester.send(commandString);
  LOGGER.debug("Command is in flight");

  String reply = command.toJson();
  Poller poller = new Poller(1);
  poller.register(requester, Poller.POLLIN);

  poller.poll(REQUEST_TIMEOUT);

  if (poller.pollin(0)) {
    reply = requester.recvStr();
    LOGGER.debug("Got response");
  }

  requester.close();

  return reply;
}
 

import org.zeromq.ZMQ.Poller; //导入依赖的package包/类
@Override
public void sendRequest(String request) throws Exception {
  int retriesLeft = REQUEST_RETRIES;

  while (retriesLeft > 0 && !Thread.currentThread().isInterrupted()) {
    LOGGER.info("Sending request " + request);
    requester.send(request);

    int expect_reply = 1;
    while (expect_reply > 0) {
      PollItem items[] = {new PollItem(requester, Poller.POLLIN)};
      int rc = ZMQ.poll(items, REQUEST_TIMEOUT);
      if (rc == -1) {
        break; //  Interrupted
      }

      if (items[0].isReadable()) {

        try {
          String response = requester.recvStr();

          retriesLeft = 0;
          expect_reply = 0;

          LOGGER.info("Handling response");
          this.handler.handle(response);
        } catch (Exception e) {
          LOGGER.warn("Error during request/response", e);

          if (e instanceof ZMQException) {
            LOGGER.error("Will close request socket");
            requester.close();
            ctx.close();
            LOGGER.info("Closed request socket, which was connected to " + responder);
            throw new Exception("Error during request/response", e);
          }
        }

      } else if (--retriesLeft == 0) {
        LOGGER.error("Service " + responder + " seems to be unavailable");
        throw new Exception("Service " + responder + " seems to be unavailable");
      } else {
        LOGGER.warn("No response from " + responder);
        requester.close();
        LOGGER.info("Reconnecting request socket to " + responder);
        connect();
      }
    }
  }
}