本文整理了Java中org.zeromq.ZMsg类的一些代码示例，展示了ZMsg类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZMsg类的具体详情如下：
包路径：org.zeromq.ZMsg
类名称：ZMsg

ZMsg介绍

[英]The ZMsg class provides methods to send and receive multipart messages across 0MQ sockets. This class provides a list-like container interface, with methods to work with the overall container. ZMsg messages are composed of zero or more ZFrame objects.

// Send a simple single-frame string message on a ZMQSocket "output" socket object 
ZMsg.newStringMsg("Hello").send(output); 
// Add several frames into one message 
ZMsg msg = new ZMsg(); 
for (int i = 0;i< 10;i++) { 
msg.addString("Frame" + i); 
} 
msg.send(output); 
// Receive message from ZMQSocket "input" socket object and iterate over frames 
ZMsg receivedMessage = ZMsg.recvMsg(input); 
for (ZFrame f : receivedMessage) { 
// Do something with frame f (of type ZFrame) 
}

Based on zmsg.c in czmq
[中]ZMsg类提供了跨0MQ套接字发送和接收多部分消息的方法。这个类提供了一个类似列表的容器接口，其中包含处理整个容器的方法。ZMsg消息由零个或多个ZFrame对象组成

// Send a simple single-frame string message on a ZMQSocket "output" socket object 
ZMsg.newStringMsg("Hello").send(output); 
// Add several frames into one message 
ZMsg msg = new ZMsg(); 
for (int i = 0;i< 10;i++) { 
msg.addString("Frame" + i); 
} 
msg.send(output); 
// Receive message from ZMQSocket "input" socket object and iterate over frames 
ZMsg receivedMessage = ZMsg.recvMsg(input); 
for (ZFrame f : receivedMessage) { 
// Do something with frame f (of type ZFrame) 
}

基于czmq中的zmsg.c

代码示例

代码示例来源：origin: twosigma/beakerx

private synchronized void sendMsg(ZMQ.Socket socket, List<Message> messages) {
 if (!isShutdown()) {
  messages.forEach(message -> {
   String header = toJson(message.getHeader());
   String parent = toJson(message.getParentHeader());
   String meta = toJson(message.getMetadata());
   String content = toJson(message.getContent());
   String digest = hmac.sign(Arrays.asList(header, parent, meta, content));
   ZMsg newZmsg = new ZMsg();
   message.getIdentities().forEach(newZmsg::add);
   newZmsg.add(DELIM);
   newZmsg.add(digest.getBytes(StandardCharsets.UTF_8));
   newZmsg.add(header.getBytes(StandardCharsets.UTF_8));
   newZmsg.add(parent.getBytes(StandardCharsets.UTF_8));
   newZmsg.add(meta.getBytes(StandardCharsets.UTF_8));
   newZmsg.add(content.getBytes(StandardCharsets.UTF_8));
   message.getBuffers().forEach(x -> newZmsg.add(x));
   newZmsg.send(socket);
  });
 }
}

代码示例来源：origin: twosigma/beakerx

private Message readMessage(ZMQ.Socket socket) {
 ZMsg zmsg = null;
 Message message = null;
 try {
  zmsg = ZMsg.recvMsg(socket);
  ZFrame[] parts = new ZFrame[zmsg.size()];
  zmsg.toArray(parts);
  byte[] uuid = parts[MessageParts.UUID].getData();
  byte[] header = parts[MessageParts.HEADER].getData();
  byte[] parent = parts[MessageParts.PARENT].getData();
  byte[] metadata = parts[MessageParts.METADATA].getData();
  byte[] content = parts[MessageParts.CONTENT].getData();
  byte[] expectedSig = parts[MessageParts.HMAC].getData();
  verifyDelim(parts[MessageParts.DELIM]);
  verifySignatures(expectedSig, header, parent, metadata, content);
  message = new Message(parse(header, Header.class));
  if (uuid != null) {
   message.getIdentities().add(uuid);
  }
  message.setParentHeader(parse(parent, Header.class));
  message.setMetadata(parse(metadata, LinkedHashMap.class));
  message.setContent(parse(content, LinkedHashMap.class));
 } finally {
  if (zmsg != null) {
   zmsg.destroy();
  }
 }
 return message;
}

代码示例来源：origin: net.unit8.enkan/enkan-repl-pseudo

@Override
public void run() {
  while(!Thread.currentThread().isInterrupted()) {
    ZMsg msg = ZMsg.recvMsg(socket);
    ZFrame clientAddress = msg.pop();
    String input = msg.popString();
    int cursor = Integer.parseInt(msg.popString());
    int[] anchor = {-1};
    ZMsg reply = new ZMsg();
    reply.add(clientAddress.duplicate());
    String trimmedCommand = input.trim();
    if (trimmedCommand.startsWith("/")) {
      if (!trimmedCommand.contains(" ")) {
        Predicate<String> filter = trimmedCommand.equals("/") ?
            n -> true : n -> n.startsWith(trimmedCommand.substring(1));
        commandNames.stream()
            .filter(filter)
            .forEach(s -> reply.add("/" + s));
        anchor[0] = 0;
      }
    }
    reply.send(socket, true);
  }
}

代码示例来源：origin: org.zeromq/jeromq

private ZMsg status()
{
  ZMsg reply = new ZMsg();
  if (!state.alive) {
    reply.add(EXITED);
  }
  else if (state.paused) {
    reply.add(PAUSED);
  }
  else if (state.started) {
    reply.add(STARTED);
  }
  else {
    reply.add(STOPPED);
  }
  return reply;
}

代码示例来源：origin: kawasima/enkan

@Override
  public int complete(String buffer, int cursor, List<CharSequence> candidates) {
    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
    ZMsg msg = new ZMsg();
    msg.add(buffer);
    msg.add(Integer.toString(cursor));
    msg.send(socket);

    ZMsg response = ZMsg.recvMsg(socket);
    while (response != null && !response.isEmpty()) {
      candidates.add(response.popString());
    }
    if (candidates.isEmpty()) return cursor;

    int delimiterPos = Math.max(buffer.lastIndexOf(' '), buffer.lastIndexOf('.'));
    if (delimiterPos > 0) {
      return delimiterPos + 1;
    } else {
      return 0;
    }
  }
}

代码示例来源：origin: kawasima/enkan

@Override
public String recv(long timeout) {
  ZMsg msg = ZMsg.recvMsg(socket);
  clientAddress = msg.pop();
  return msg.popString();
}

代码示例来源：origin: org.zeromq/jeromq

protected ZAuth send(String command, String... datas)
{
  ZMsg msg = new ZMsg();
  msg.add(command);
  for (String data : datas) {
    msg.add(data);
  }
  agent.send(msg);
  msg.destroy();
  agent.recv();
  return this;
}

代码示例来源：origin: org.zeromq/jeromq

@Override
  public boolean flow(Plug splug, Socket source, Socket capture, Plug dplug, Socket destination)
  {
    boolean success;
    // we read the whole message
    ZMsg msg = ZMsg.recvMsg(source);
    if (msg == null) {
      return false;
    }
    if (capture != null) {
      //  Copy transformed message to capture socket if any message
      // TODO what if the transformer modifies or destroys the original message ?
      ZMsg cpt = transformer.transform(msg, splug, Plug.CAPTURE);
      //                boolean destroy = !msg.equals(cpt); // TODO ?? which one
      boolean destroy = msg != cpt;
      success = cpt.send(capture, destroy);
      if (!success) {
        // not successful, but we can still try to send it to the destination
      }
    }
    ZMsg dst = transformer.transform(msg, splug, dplug);
    // we send the whole transformed message
    success = dst.send(destination);
    // finished
    msg.destroy();
    return success;
  }
}

代码示例来源：origin: org.projectreactor/reactor-net

@Override
  public int handle(ZLoop loop, ZMQ.PollItem item, Object arg) {
    ZMsg msg = ZMsg.recvMsg(socket);
    if (null == msg || msg.size() == 0) {
      return 0;
    }
    if (closed) {
      return -1;
    }
    String connId;
    switch (ZeroMQWorker.this.socketType) {
      case ZMQ.ROUTER:
        connId = msg.popString();
        break;
      default:
        connId = ZeroMQWorker.this.id.toString();
    }
    ZeroMQNetChannel<IN, OUT> netChannel = select(connId)
        .setConnectionId(connId)
        .setSocket(socket);
    ZFrame content;
    while (null != (content = msg.pop())) {
      netChannel.read(Buffer.wrap(content.getData()));
    }
    msg.destroy();
    return 0;
  }
};

代码示例来源：origin: org.zeromq/jeromq

/**
 * Send message to 0MQ socket.
 *
 * @param socket
 *              0MQ socket to send ZMsg on.
 * @return true if send is success, false otherwise
 */
public boolean send(Socket socket)
{
  return send(socket, true);
}

代码示例来源：origin: org.zeromq/jeromq

/**
 * Receives message from socket, returns ZMsg object or null if the
 * recv was interrupted. Does a blocking recv, if you want not to block then use
 * the ZLoop class or ZMQ.Poller to check for socket input before receiving or recvMsg with flag ZMQ.DONTWAIT.
 * @param   socket
 * @return
 *          ZMsg object, null if interrupted
 */
public static ZMsg recvMsg(Socket socket)
{
  return recvMsg(socket, 0);
}

代码示例来源：origin: org.opendaylight.netide/shim

poller.poll(10);
if (poller.pollin(0)) {
  ZMsg message = ZMsg.recvMsg(socket);
  byte[] data = message.getLast().getData();
  if (coreListener != null) {
    Message msg = NetIPConverter.parseConcreteMessage(data);
  ZMsg message = ZMsg.recvMsg(controlSocket);
  if (message.getFirst().toString().equals(STOP_COMMAND)) {
    break;
  } else {
    message.send(socket);

代码示例来源：origin: org.zeromq/jeromq

/**
 * Restarts the proxy. Stays alive.
 *
 * @param hot     null to make a cold restart (closing then re-creation of the sockets)
 *  or a configuration message to perform a configurable hot restart,
 */
public String restart(ZMsg hot)
{
  ZMsg msg = new ZMsg();
  msg.add(RESTART);
  final boolean cold = hot == null;
  if (cold) {
    msg.add(Boolean.toString(false));
  }
  else {
    msg.add(Boolean.toString(true));
    msg.append(hot);
  }
  String status = EXITED;
  if (agent.send(msg)) {
    status = status(false);
  }
  return status;
}

代码示例来源：origin: org.zeromq/jeromq

/**
   * Send a zap reply to the handler socket
   */
  private void reply(int statusCode, String statusText, Socket replies)
  {
    ZapReply reply = new ZapReply(ZAP_VERSION, sequence, statusCode, statusText, userId, metadata);
    ZMsg msg = reply.msg();
    boolean destroy = replies == null;
    msg.send(handler, destroy);
    if (replies != null) {
      // let's add other fields for convenience of listener
      msg.add(address);
      msg.add(identity);
      msg.send(replies);
    }
  }
}

代码示例来源：origin: org.zeromq/jeromq

@Override
public boolean backstage(Socket pipe, ZPoller poller, int events)
  ZMsg msg = ZMsg.recvMsg(pipe);
  String command = msg.popString();
  if (command == null) {
    System.out.printf("ZAuth: Closing auth: No command%n");
    String address = msg.popString();
    if (verbose) {
      System.out.printf("ZAuth: Whitelisting IP address=%s\n", address);
    String address = msg.popString();
    if (verbose) {
      System.out.printf("ZAuth: Blacklisting IP address=%s\n", address);
    String verboseStr = msg.popString();
    this.verbose = Boolean.parseBoolean(verboseStr);
    rc = pipe.send(OK);
    repliesEnabled = Boolean.parseBoolean(msg.popString());
    if (verbose) {
      if (repliesEnabled) {
  msg.destroy();
  if (!rc) {
    System.out.printf("ZAuth: Command in error %s%n", command);

代码示例来源：origin: org.zeromq/jeromq

public void append(ZMsg msg)
{
  if (msg == null) {
    return;
  }
  for (ZFrame frame : msg.frames) {
    add(frame);
  }
}

代码示例来源：origin: org.zeromq/jeromq

private static ZapReply received(ZMsg msg)
  {
    if (msg == null) {
      return null;
    }
    assert (msg.size() == 8);
    String version = msg.popString();
    String sequence = msg.popString();
    int statusCode = Integer.parseInt(msg.popString());
    String statusText = msg.popString();
    String userId = msg.popString();
    ZMetadata metadata = ZMetadata.read(msg.popString());
    String address = msg.popString();
    String identity = msg.popString();
    return new ZapReply(version, sequence, statusCode, statusText, userId, metadata, address, identity);
  }
}

代码示例来源：origin: org.zeromq/jeromq

@Override
public ZMsg recv(boolean wait)
{
  if (locked) {
    return null;
  }
  try {
    ZMsg msg = ZMsg.recvMsg(pipe, wait);
    if (msg == null) {
      return null;
    }
    if (msg.size() == 1) {
      final ZFrame frame = msg.peek();
      byte[] key = frame.getData();
      if (lock != null && Arrays.equals(lock, key)) {
        locked = true;
        // this is the last message anyway, and not a one for a public display
        msg = null;
        pipe.close();
      }
    }
    return msg;
  }
  catch (ZMQException e) {
    locked = true;
    return null;
  }
}

代码示例来源：origin: io.projectreactor/reactor-net

@Override
  public int handle(ZLoop loop, ZMQ.PollItem item, Object arg) {
    ZMsg msg = ZMsg.recvMsg(socket);
    if (null == msg || msg.size() == 0) {
      return 0;
    }
    if (closed) {
      return -1;
    }
    b.onNext(msg);
    return 0;
  }
};

代码示例来源：origin: kawasima/enkan

socket.monitor("inproc://socket.monitor", ZMQ.EVENT_DISCONNECTED);
ZMsg completerMsg = ZMsg.recvMsg(socket);
ReplResponse completerRes = fressian.read(completerMsg.pop().getData(), ReplResponse.class);
String completerPort = completerRes.getOut();
if (completerPort != null && completerPort.matches("\\d+")) {
  while (true) {
    try {
      ZMsg msg = ZMsg.recvMsg(this.socket);
      ReplResponse res = fressian.read(msg.pop().getData(), ReplResponse.class);
      if (res.getOut() != null) {
        console.println(res.getOut());