使用redis的queue-outbound适配器分发任务丢失消息答案

【问题标题】：Missing messages using queue-outbound adapter of redis to distribute task使用redis的queue-outbound适配器分发任务丢失消息
【发布时间】：2014-07-03 15:26:47
【问题描述】：

我正在使用 s.i. 提供的两个组件。有一个有效的分配系统

消息（任务）被发送到这个名为分发的通道；分发有一个与通知系统关联的窃听器，因此当消息通过分发时，我可以通过 jconsole 或任务控制看到：

  <int:channel id="distribution">
        <int:interceptors>
            <int:wire-tap channel="distributionPublish"/>
        </int:interceptors>
    </int:channel>

那我用redis作为队列系统：

  <redis:queue-outbound-channel-adapter
            id="toRedis" channel="distribution" queue="Qname"
            auto-startup="true" extract-payload="false"  />


    <!-- a Queue Inbound Channel Adapter is available to 'right pop' messages
        from a Redis List. -->
    <redis:queue-inbound-channel-adapter
            id="fromRedis" channel="execution" queue="Qname"
        receive-timeout="1000" recovery-interval="1000" expect-message="true"
            auto-startup="true"/>

使用以下配置和 2 个服务器，每台接收 50% 的消息；主要是一台服务器有web服务接口，将消息放入输入队列，然后所有订阅和阻塞的服务器都在获取消息（任务）

但是当我加快系统速度时，会丢失消息。我不知道为什么，也不知道我能做什么。我已经增加了 redis 池的值，这个问题还在继续，

我是不是做错了什么，或者如何在分发组件中实现“重试”？

PD：我通过窃听来确定错误是在分发组件中

更新

也许这个配置有帮助？（还没有测试，我会更新）这个想法是让几个线程去redis

  <int:channel id="distribution">
        <int:dispatcher task-executor="DistributionTaskExecutor"/>
        <int:interceptors>
            <int:wire-tap channel="distributionPublish"/>
        </int:interceptors>
    </int:channel>
    <!-- to handle high demanding we use several threads to go to redis -->
    <task:executor id="DistributionTaskExecutor" pool-size="2" />

    <redis:queue-outbound-channel-adapter
            id="toRedis" channel="distribution" queue="${instance}"
            auto-startup="true" extract-payload="false"  />

更新：

最后，我正在使用此配置，并且似乎运行良好。至少，更好：

 <int:channel id="distribution">
        <int:queue capacity="50"/>
    </int:channel>

    <task:executor id='distributionExecutor' pool-size='25' queue-capacity='25' rejection-policy="CALLER_RUNS"/>

    <redis:queue-outbound-channel-adapter
            id="toRedis" channel="distribution" queue="${instance}"
            auto-startup="true" extract-payload="false">
        <int:poller task-executor='distributionExecutor' fixed-delay='500'>
        </int:poller>
    </redis:queue-outbound-channel-adapter>

【问题讨论】：

如果您只有一个订阅服务器，您是否会丢失消息？
我已经编辑了帖子试图更好地解释；...我迷失在“分发”（消息实际上没有发送到队列）而不是接收中。或者这是我的嫌疑人......
操作！在分发之前，我有一个带有 resolution-required=true 的 header-value-router .... 也许我错过了这个组件中的消息。在我正在阅读的文档中，使用默认输出通道所需的分辨率应该是错误的..
???它可以解决您的问题吗？所以你只是没有显示整个配置
嗯，这只能在生产和高要求的峰值中重现，所以我现在无法测试。当 redis 队列的 2 个消费者竞争正确弹出消息时，消息是否可能丢失（对于您的问题）？

标签： redis spring-integration

【解决方案1】：

为了完成这个问题，我将我的 cmets 移到这里作为答案。

由于<task:executor/> 配置具有非常低的并发性 (2) 并且使用AbortPolicy 作为默认配置，因此由于rejection 而对 Redis 执行的任务已经丢失。

只需要依赖单线程DirectChannel 或确保任务执行器有足够大的线程池，并为CallerRunsPolicy 配置策略。

【讨论】：