java - JeroMQ 不时产生“IOException。打开的文件太多”

Question

我有一个 tomcat7 应用程序，它使用 IPC 端点和 Java 中的 JeroMQ 与我的另一个应用程序通信。有一个客户端服务器方案，客户端有时会等待来自服务器的响应，如果它没有收到响应，则第一次失败而不重试。

代码如下

@Override
public List<Result> call() throws Exception {
    final List<Result> results = new LinkedList<>();
    try {
        for (DTO dto : messages) {
            Message m = MessageHelper.MessageMapper(dto);

            Thread.sleep(dto.getDelayBeforeSend());
            final Result mtresult = send(dto);
            results.add(result);
        }
    } catch (RuntimeException e) {
        LOGGER.error("Flow => Uncaught Exception: {}", e.getMessage());
        LOGGER.debug("Flow => Uncaught Exception: ", e);
        Thread t = Thread.currentThread();
        t.getUncaughtExceptionHandler().uncaughtException(t, e);
    }
    return results;
}

private Result send(Message m) {
    ZMQ.Socket client = MQSocketFactory.getMQSocket(serverEndpoint).createRequester();
    try {
        final byte[] DTO = Helper.serializeMessage(m);
        int retriesLeft = 1;
        Result result = new Result(MessageConstants.MESSAGE_FAIL);

        while (retriesLeft > 0 && !Thread.currentThread().isInterrupted()) {

            client.send(DTO, 0);
            int expect_reply = 1;

            while (expect_reply > 0) {

                ZMQ.PollItem items[] = { new ZMQ.PollItem(client, Poller.POLLIN) };
                int rc = ZMQ.poll(items, 3000);
                if (rc == -1) break; // Interrupted

                if (items[0].isReadable()) {
                    final byte[] reply = client.recv(0);
                    if (reply == null) break;
                    result = new Result(new String(reply));
                    if (result.isSuccessful()) {
                        LOGGER.trace("Server replied OK. Result: [{}]", result);
                        retriesLeft = 0;
                        expect_reply = 0;
                    } else LOGGER.error("Malformed reply from server: [{}]", result);

                } else if (--retriesLeft == 0) {
                    LOGGER.error("Server:[{}] seems to be offline, abandoning sending message [{}]!", serverEndpoint, m);
                    break;
                } else {
                    LOGGER.warn("No response from server, retrying...");
                    client = MQSocketFactory.getMQSocket(serverEndpoint).resetRequester(client);
                    client.send(DTO, 0);
                }
            }
        }
        return result;
    } finally {
        MQSocketFactory.getMQSocket(serverEndpoint).destroyRequester(client);

    }
}

现在 MQSocketFactory 类如下所示：

public final class MQSocketFactory {

private static final Map<String, MQSocket> store = new HashMap<String, MQSocket>();

private static final Logger LOGGER = LoggerFactory.getLogger(MQSocketFactory.class);

public static MQSocket getMQSocket(String endpointName) {
    synchronized (store) {
        MQSocket result = store.get(endpointName);
        if (result == null) {
            result = new MQSocket(endpointName);
            store.put(endpointName, result);
        }
        return result;
    }
}

public static final class MQSocket {

    private final String endpoint;
    private final ZMQ.Context ctx;

    private MQSocket(String endpointName) {
        this.endpoint = endpointName;
        this.ctx = ZMQ.context(1);
    }

    public ZMQ.Socket createRequester() {
        ZMQ.Socket client = null;
        try {
            client = ctx.socket(ZMQ.REQ);
            assert (client != null);
            client.connect(endpoint);
        } catch (Exception e) {
            LOGGER.error("Error: {}", e.getMessage());
            LOGGER.error("Error: {}", e);
        }
        return client;
    }

    public ZMQ.Socket resetRequester(ZMQ.Socket socket) {
        destroyRequester(socket);
        return createRequester();
    }

    public void destroyRequester(ZMQ.Socket socket) {
        if (socket != null) {
            try {
                socket.close();
            } catch (Exception e) {
                LOGGER.error("Error: {}", e.getMessage());
                LOGGER.debug("Error: {}", e);
            }
        }
    }

    public ZMQ.Context getContext() {
        return ctx;
    }

    // Responder Unit
    private ZMQ.Socket responder;

    public ZMQ.Socket createResponder() {
        if (responder == null) {
            this.responder = ctx.socket(ZMQ.REP);
            responder.bind(endpoint);
        }
        return responder;
    }

    public ZMQ.Socket resetResponder() {
        destroyResponder();
        return createResponder();
    }

    public void destroyResponder() {
        try {
            responder.close();
        } catch (Exception e) {
            LOGGER.error("Error: {}", e.getMessage());
            LOGGER.debug("Error: {}", e);
        }
    }

}

}

我专门这样做了，所以每个套接字在请求完成后都会关闭，以避免 IOExcpetion Too Many file Open 出现这个特殊问题。但是，我很少遇到这个问题，我不知道为什么。该应用程序可能在几乎相同的负载下工作了几天，一切正常，但在某些时候它开始抛出异常，我不知道为什么。

还有没有办法增加tomcat7中的ulimit？现在是1024。