如何发送消息以停止轮询并退出应用程序

Question

我有 spring 集成应用程序，我需要在处理完所有数据后关闭它。如果我明确调用appContext.close()，那么并非所有数据都可以及时处理（除非我设置Thread.sleep()）。如果我不在应用程序上下文上调用 close ，那么应用程序不会停止，因为我有不允许应用程序自动关闭的后台轮询器。那么如何在我的一个服务激活器（处理链中的最后一个）中发出停止整个应用程序的信号？

1. 第一个 bean 从存储中逐行读取数据，并在 while 循环中通过gateway.send(data) 发送
2. 并行发生的处理链
3. 所有线程都通过 Pollable Queue 将消息发送到单个线程
4. 如果我意识到在第一个 bean 中读取的所有消息都已处理，我应该在此处停止应用程序

我尝试使用 controlbus 停止最后一个服务激活器，但没有帮助

谢谢

更新

这里有一些代码示例：

跑步者：

public class Runner {
static Logger log = LoggerFactory.getLogger(Service2.class);

public static void main(String[] args) throws InterruptedException {
    log.info("START APP");
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(AppConfig.class);
    RootService service = context.getBean(RootService.class);
    service.start();
    service.stop();
    context.close();
    log.info("END APP");
}

}

根服务：

    @Component
public class RootService {

    Logger log = LoggerFactory.getLogger(RootService.class);

    @Autowired
    MyGateway gateway;

    int totalSize = 0;

    public void start() {
        List<String> source = generateSource();
        totalSize = source.size();
        //imitate very long but finite process
        for (String s : source) {
            gateway.send(s, totalSize);
        }
        log.info("end sending data");
    }

    public void stop() throws InterruptedException {
        log.info("sending stop signal...");
        while (gateway.sendStop(totalSize)<0) {
            Thread.sleep(100);
            log.info("sending stop signal...");
        }
        log.info("THE END");
    }


    private List<String> generateSource() {
        List<String> result = new ArrayList<String>();
        for (int i = 0; i < 15; i++) {
            result.add("data" + i);
        }
        return result;
    }
}

服务1

@Component
public class Service1 {

    public String dodo(String data) throws InterruptedException {
        //doing a job in parallel
        Thread.sleep(100);
        return data + "-" + Thread.currentThread().getName();
    }
}

服务2：

    @Component
public class Service2 {
    Logger log = LoggerFactory.getLogger(Service2.class);
    int counter = 0;

    public void dodo(String data) {
        log.info("data: {}-{}", data, Thread.currentThread().getName());
        counter++;
        log.info("counter: {}", counter);
    }

    public Integer dodo(Integer data) {
        if (counter < data) {
            return -1;
        } else {
            return 0;
        }
    }
}


@Component
public class ErrorHandler {

    Logger log = LoggerFactory.getLogger(Service2.class);

    public void handleError(Message<?> message) {
        log.info("ERROR: {}", message);
    }
}

和xml配置

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"
             xmlns:beans="http://www.springframework.org/schema/beans"
             xmlns:task="http://www.springframework.org/schema/task">


    <gateway id="myGateway"
             service-interface="com.dimas.MyGateway"
             default-request-channel="channel1"
             error-channel="errorChannel"
             default-reply-timeout="3000">
        <method name="send" request-channel="channel1"/>
        <method name="sendStop" request-channel="channel2" reply-channel="channel3"/>
    </gateway>

    <channel id="channel1">
        <dispatcher task-executor="executor"/>
    </channel>
    <channel id="channel2">
        <queue/>
    </channel>
    <channel id="channel3">
        <queue/>
    </channel>
    <channel id="errorChannel"/>

    <service-activator input-channel="errorChannel" ref="errorHandler" method="handleError"/>

    <service-activator input-channel="channel1" output-channel="channel2" ref="service1"/>
    <service-activator input-channel="channel2" output-channel="channel3" ref="service2">
        <poller fixed-delay="0"/>
    </service-activator>

    <task:executor id="executor" pool-size="2"/>
</beans:beans>

现在它可以工作了 - 处理完所有数据后应用程序停止并返回停止代码 0，但我在日志中看到很多错误，例如：

错误消息 [有效负载=org.springframework.messaging.core.DestinationResolutionException： 没有可用的 output-channel 或 replyChannel 标头， 标头={id=639ca939-8110-4486-6a2b-5d36c7bfdbcd, 时间戳=1475872251269}]

错误处理程序成功捕获它但有问题

我意识到问题出在哪里。最后一个 Service2 为任何收入消息返回一些东西。重做它，使其仅响应 stopRequest 请求的代码。

只是想知道是否有更简单的解决方案

Answer 1

不清楚为什么，如果你能检测到你是完整的，你就不能关闭应用程序上下文。

您可以将默认的taskScheduler 替换为使用守护线程的@。

编辑

首先，您不需要所有这些队列通道；简单地让 channel1 成为执行者通道会给你带来并发性 - 之后使用队列通道只会增加开销。

无论如何，要确定该过程何时完成，只需在Service2中添加一个getCounter()方法即可；然后，在您的 main 方法中，从上下文中获取 service2 并等待计数器增加到您期望的数字。

或者，您可以向 service2 添加一个倒计时锁存器 - 添加一个方法来设置它...

CountDownLatch latch = new CountDownLatch(source.size());
service2.setCountDownLatch(latch);
for (

...
if (!latch.await(...)) { // add a timeout in case is never completes)
    // failure
}