如何让Java应用通过docker-java等待容器内长命令执行完成？

问题描述 #

我有一个Docker Swarm栈，包含多个服务，其中一个Java应用容器需要在另一个容器的shell中执行命令。两个服务部署在同一机器上，因此应用可通过docker-java库发起命令。执行mkdir、rm等瞬时命令时一切正常，但实际需执行的是数据处理类长命令（耗时不低于数分钟）。Java应用需等待该命令执行完成后，再清理目标服务要处理的数据，但目前代码无法实现同步，会提前删除数据。仅靠Thread.sleep固定等待可行，但因处理时长不固定需移除该逻辑。

当前代码示例：

String dockerSocket = "unix:///var/run/docker.sock";

DockerClientConfig config = DefaultDockerClientConfig.createDefaultConfigBuilder()
                .withDockerHost(dockerSocket)
                .build();

DockerClient dockerClient = DockerClientBuilder.getInstance(config).build();

List<Container> containers = dockerClient.listContainersCmd().withNameFilter(Collections.singleton("accumulo-master")).exec();

try {
    logger.info("Trying to execute ingestion command");

    if (!containers.isEmpty()) {
        String containerID = containers.get(0).getId();
        logger.info("containerID: " + containerID);

        String[] command = { "/bin/bash", "-c", "command to execute" };

        ExecCreateCmdResponse execCreateCmdResponse = dockerClient.execCreateCmd(containerID)
                .withCmd(command)
                .exec();

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

        ExecStartResultCallback execStartResultCallback = new ExecStartResultCallback(outputStream, System.err) {
            @Override
            public void onComplete() {
                logger.info("Results");
                logger.info(outputStream.toString());
                try {
                    Thread.sleep(120000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        };

        dockerClient.execStartCmd(execCreateCmdResponse.getId())
                .exec(execStartResultCallback).onComplete();
    }
    // 此处若有清理数据的逻辑，会提前执行
} catch (Exception e) {
    logger.error("Error executing command", e);
}

解决方案 #

方式1：将清理逻辑移至回调的onComplete()方法 #

ExecStartResultCallback的onComplete()方法会在容器内命令执行完成后才被调用，因此只需把清理数据的代码放到这个方法里，就能保证命令执行完毕后再执行清理：

String dockerSocket = "unix:///var/run/docker.sock";

DockerClientConfig config = DefaultDockerClientConfig.createDefaultConfigBuilder()
                .withDockerHost(dockerSocket)
                .build();

DockerClient dockerClient = DockerClientBuilder.getInstance(config).build();

List<Container> containers = dockerClient.listContainersCmd().withNameFilter(Collections.singleton("accumulo-master")).exec();

try {
    logger.info("Trying to execute ingestion command");

    if (!containers.isEmpty()) {
        String containerID = containers.get(0).getId();
        logger.info("containerID: " + containerID);

        String[] command = { "/bin/bash", "-c", "command to execute" };

        ExecCreateCmdResponse execCreateCmdResponse = dockerClient.execCreateCmd(containerID)
                .withCmd(command)
                .exec();

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

        ExecStartResultCallback execStartResultCallback = new ExecStartResultCallback(outputStream, System.err) {
            @Override
            public void onComplete() {
                logger.info("Results");
                logger.info(outputStream.toString());
                // 在这里执行清理数据的逻辑
                logger.info("开始清理目标服务数据");
                // 调用你的清理方法，比如cleanupTargetData();
            }

            @Override
            public void onError(Throwable throwable) {
                logger.error("命令执行失败", throwable);
                // 可选：命令执行失败时的处理逻辑，比如不清理或标记错误
            }
        };

        dockerClient.execStartCmd(execCreateCmdResponse.getId())
                .exec(execStartResultCallback);
    }
} catch (Exception e) {
    logger.error("Error executing command", e);
}

方式2：用CountDownLatch实现主线程同步等待 #

如果清理逻辑必须在主线程中执行，可以通过CountDownLatch阻塞主线程，直到命令执行完成：

import java.util.concurrent.CountDownLatch;

String dockerSocket = "unix:///var/run/docker.sock";

DockerClientConfig config = DefaultDockerClientConfig.createDefaultConfigBuilder()
                .withDockerHost(dockerSocket)
                .build();

DockerClient dockerClient = DockerClientBuilder.getInstance(config).build();

List<Container> containers = dockerClient.listContainersCmd().withNameFilter(Collections.singleton("accumulo-master")).exec();

try {
    logger.info("Trying to execute ingestion command");

    if (!containers.isEmpty()) {
        String containerID = containers.get(0).getId();
        logger.info("containerID: " + containerID);

        String[] command = { "/bin/bash", "-c", "command to execute" };

        ExecCreateCmdResponse execCreateCmdResponse = dockerClient.execCreateCmd(containerID)
                .withCmd(command)
                .exec();

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        CountDownLatch latch = new CountDownLatch(1); // 初始化计数器为1

        ExecStartResultCallback execStartResultCallback = new ExecStartResultCallback(outputStream, System.err) {
            @Override
            public void onComplete() {
                logger.info("Results");
                logger.info(outputStream.toString());
                latch.countDown(); // 命令完成，计数器减1
            }

            @Override
            public void onError(Throwable throwable) {
                logger.error("命令执行失败", throwable);
                latch.countDown(); // 执行失败也释放计数器，避免主线程永久阻塞
            }
        };

        dockerClient.execStartCmd(execCreateCmdResponse.getId())
                .exec(execStartResultCallback);

        // 主线程阻塞，直到计数器变为0
        latch.await();
        // 命令执行完成（或失败）后，执行清理逻辑
        logger.info("开始清理目标服务数据");
        // cleanupTargetData();
    }
} catch (Exception e) {
    logger.error("Error executing command", e);
}

关键说明 #

• docker-java的execStartCmd默认是异步执行的，主线程不会等待命令完成，这是导致提前清理数据的根本原因。
• 两种方式都不需要Thread.sleep，能根据命令实际执行时长自动等待，避免了固定等待时间的局限性。
• 方式1更符合异步编程的设计，方式2适合必须在主线程中执行后续逻辑的场景。

内容的提问来源于stack exchange，提问作者Luigi