使用Camel RoutePolicy设置sqs的高优先级/低优先级路由使用者的优雅方法？

Question

我有两个SQS队列：一个用于低优先级，另一个用于高优先级消息。逻辑是不要接触低优先级队列上的消息，除非高优先级队列为空。

SuspendLowPriorityRoutePolicy suspendLowPriorityRoutePolicy = new SuspendLowPriorityRoutePolicy(LOW_PRIORITY_ROUTE_ID, camelContext);

        from(UriBuilder.buildSqsUri(sqsProperties)).routeId(LOW_PRIORITY_ROUTE_ID)
                .log(LoggingLevel.INFO, log, "NON-PRIORITY: ${body}");

        from(UriBuilder.buildPrioritySqsUri(sqsProperties)).routeId(HIGH_PRIORITY_ROUTE_ID)
                .routePolicy(suspendLowPriorityRoutePolicy)
                .log(LoggingLevel.INFO, log, "PRIORITY: ${body}");

现在，我将这两条路由设置为同时使用队列中的消息。我想要的是，一个消息进入高优先级路由触发低优先级路由的停止。为了尝试获得此功能,我尝试使用一种路由策略,当在高优先级路由上启动新交换时,该策略将停止低优先级队列：

（来自SuspendLowPriorityRoutePolicy的片段）

 @Override
    public void onExchangeBegin(Route route, Exchange exchange) {
        Route lowPriorityRoute = context.getRoute(lowPriorityRouteId);
        ServiceStatus routeStatus = context.getRouteStatus(lowPriorityRouteId);
        if (!routeStatus.isStopped()) {
            try {
                lock.lock();
                log.info("High priority request came in, stopping consumer");
                stopConsumer(lowPriorityRoute.getConsumer());
            } catch (Exception e) {
                log.error("Exception stopping consumer " + e);
                handleException(e);
            } finally {
                lock.unlock();
            }
        }
    }

然而，我不确定如何重新启动低优先级的消费者。骆驼routepolicy提供的其他钩子允许重写onExchangeDone,但此时逻辑应该是只有当高优先级队列为空时才重新启动低优先级使用者。我认为没有一种方法可以检查队列是否为空，我们可以检查exchange done钩子上的ExampiateNumberOfMessages属性，但这可能是不准确的。

另一种想法是有一个调度的后台轮询器，该轮询器检查CamelContext的inFlightRequestsRepository，并且只有在高优先级路由没有飞行请求时才重新启动低优先级队列。

Answer 1

好吧，我能想到的最简单的方法是，用比低优先级消息多得多的消费者来消费高优先级消息。这样，高优先级的消息通常会很快被使用，而低优先级的消息可能会堆积，因此必须等待。

然而，低优先级消费者仍然会处理消息，无论同时处理的高优先级消息有多少。

更接近您的用例的一点是使用JMS优先级。

不过，有一些事情要记住：

• 使用消息优先级根据需要重新排序队列中的消息。这会降低包含大量消息的队列的性能。
• 消息优先级可能必须在代理或队列上激活。否则，轻重缓急就会被忽略。与您的代理一起搜索优先级示例以检查此内容。
• 我不确定，但我认为JMS连接上也有一些标志来尊重使用者的优先级
• 您必须保持使用者预取为低电平。如果您的使用者预取了1000条低优先级的消息，它会在执行其他操作之前处理所有这些消息。无论同时是否有高优先级消息到达队列。预取越大，在处理新到达的高prio消息之前，可以处理更多的低prio消息。