HAPI - 如何正确停止 SimpleServer 并防止进一步连接

Question

我正在构建一个应用程序，其中包含多个由 CommunicationProcess 类管理的服务器和客户端 HL7 连接。应用程序的部分功能是在添加新连接时重新启动该过程。客户端连接不会造成问题，因为一旦客户端停止，服务器端就无法重新连接。但是，对于服务器连接，我似乎从（相当激进的）客户端立即重新连接。这是我必须停止服务器连接的代码：

public void disconnect() 
{
    usageServer.getRemoteConnections().forEach((connection) -> connection.close());
    usageServer.stopAndWait();
    usageServer.getRemoteConnections().forEach((connection) -> connection.close());   
}

public void stop()
{
    running.set(false);

    disconnect();
}

这是我的 connectionReceived 实现：

@Override
public void connectionReceived(Connection theC) 
{
    if (running.get())
    {
        setStatus(ConnectionStatus.CONNECTED);
    }
    else
    {
        theC.close();
    }
}

如您所见，这个想法是在接收到来自 CommunicationProcess 类的停止信号时将全局 AtomicBoolean 设置为 false，该信号拒绝任何新连接，并停止服务器。不知何故，这仍然允许客户端在此过程中保持连接。客户端是一个我不能命名的应用程序，但它已经存在了十多年，我知道事实上这不会成为问题，因为多年来我一直在支持它作为我日常工作的一部分它根本不像那样。

知道为什么我的代码实际上并没有终止连接吗？我觉得我已经探索了很多这个 API，但我没有找到一种方法来取消注册一个可能会解决这个问题的连接侦听器。此外，我看不到扩展这些服务器类的方法，因为所有东西都被相当凶猛地封装和私有化了。

谢谢

Answer 1

我正在查看 HAPI 库的代码。

您描述的行为的原因可能如下。

当服务器启动时，它们会创建一个名为 AcceptorThread 的组件。顾名思义，这个线程的职责是初始化ServerSocket，用于接收传入的客户端连接，并接受它们。

这个线程，就像 API 提出的每个 Service 抽象一样，在这样的循环中运行：

/**
  * Runs the thread.
  * 
  * @see java.lang.Runnable#run()
  */
public final void run() {
  try {
    afterStartup();
    log.debug("Thread {} entering main loop", name);
    while (isRunning()) {
      handle();
      startupLatch.countDown();
    }
    log.debug("Thread {} leaving main loop", name);
  } catch (RuntimeException t) {
    if (t.getCause() != null) {
      serviceExitedWithException = t.getCause();
    } else {
      serviceExitedWithException = t;
    }
    log.warn("Thread exiting main loop due to exception:", t);
  } catch (Throwable t) {
    serviceExitedWithException = t;
    log.warn("Thread exiting main loop due to exception:", t);
  } finally {
    startupLatch.countDown();
    afterTermination();
  }

}

当您在服务器中调用方法stopAndWait 时，它也会尝试停止该线程。

停止过程基本上改变了boolean标志，该标志控制组件是否“ìsRunning()”。

如您所见，虽然它设置了标志为false，但循环中方法handle的调用仍然必须结束。

这是AcceptorThreadhandle方法的实现：

@Override
protected void handle() {
  try {
    Socket s = ss.accept();
    socketFactory.configureNewAcceptedSocket(s);
    if (!queue.offer(new AcceptedSocket(s))) {
      log.error("Denied enqueuing server-side socket {}", s);
      s.close();
    } else
      log.debug("Enqueued server-side socket {}", s);
  } catch (SocketTimeoutException e) { /* OK - just timed out */
    log.trace("No connection established while waiting");
  } catch (IOException e) {
    log.error("Error while accepting connections", e);
  }
}

如您所见，该方法调用ServerSocket.accept，从而允许新的传入连接。

为了断开这个服务器端套接字，我们可以从另一个线程调用close。

其实这个过程就是AcceptorTreadafterTermination方法实现的：

@Override
protected void afterTermination() {
  try {
    if (ss != null && !ss.isClosed())
      ss.close();
  } catch (IOException e) {
    log.warn("Error during stopping the thread", e);
  }
}

不幸的是 - 你是对的，API 非常接近！ - 没有明确的方法来做到这一点。

一种可能的解决方案是实现您自己的HL7Service，将其命名为MySimpleServer，使用SimpleServer 的代码作为基线，并仅更改方法afterTermination 的实现：

/**
  * Close down socket
  */
@Override
protected void afterTermination() {
  super.afterTermination();
  // Terminate server side socket
  acceptor.afterTermination();
  // Terminate the acceptor thread itself
  acceptor.close();
}

请注意：不是调用acceptor.stop()，而是调用acceptor.afterTermination()直接关闭底层服务器端套接字。

为避免AcceptorThread 中handle 方法引发的错误，我们还可以从原来的类中实现一个新类，或者只是尝试覆盖handle 方法以考虑是否服务器端套接字已关闭：

@Override
protected void handle() {
  try {
    if (ss.isClosed()) {
      log.debug("The server-side socket is closed. No new connections will be allowed.");
      return;
    }

    Socket s = ss.accept();
    socketFactory.configureNewAcceptedSocket(s);
    if (!queue.offer(new AcceptedSocket(s))) {
      log.error("Denied enqueuing server-side socket {}", s);
      s.close();
    } else
      log.debug("Enqueued server-side socket {}", s);
  } catch (SocketTimeoutException e) { /* OK - just timed out */
    log.trace("No connection established while waiting");
  } catch (IOException e) {
    log.error("Error while accepting connections", e);
  }
}

为了测试，你可以试试这样的：

public static void main(String[] args) throws Exception {

  HapiContext ctx = new DefaultHapiContext();

  HL7Service server = new MySimpleServer(8888);
  server.startAndWait();

  Connection client1 = ctx.newClient("127.0.0.1", 8888, false);

  server.getRemoteConnections().forEach((connection) -> connection.close());

  server.stopAndWait();

  try {
    Connection client2 = ctx.newClient("127.0.0.1", 8888, false);
  } catch (Throwable t) {
    t.printStackTrace();
  }

  ctx.close();

  System.exit(0);
}