谷歌待了几天,我相信我完全迷失了 . 我想实现一种有大约3个队列的优先级队列:
-
高优先级队列(每日),需要先进行处理 .
-
中间优先级队列(每周),如果队列#1中没有项目将处理 . (这个队列中的消息是好的,它根本不会处理)
-
低优先级队列(每月),如果队列#1和#2中没有项目将处理 . (这个队列中的消息是好的,它根本不会处理)
最初,我有以下流程,让消费者使用来自所有三个队列的消息,并检查队列#1,#2和#3中是否有任何项目 . 然后我意识到这是错误的,因为:
-
我完全迷失了一个问题:"How do I know which queue it is coming from?" .
-
我已经在消耗任何队列中的消息了,所以如果我从较低优先级队列中获取一个对象,如果我发现优先级较高的队列中有消息,我是否会把它放回队列?
以下是我目前的配置,它显示了我是个白痴 .
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<rabbit:connection-factory id="connectionFactory" host="localhost" />
<rabbit:template id="amqpTemplatead_daily" connection-factory="connectionFactory"
exchange="" routing-key="daily_queue"/>
<rabbit:template id="amqpTemplatead_weekly" connection-factory="connectionFactory"
exchange="" routing-key="weekly_queue"/>
<rabbit:template id="amqpTemplatead_monthly" connection-factory="connectionFactory"
exchange="" routing-key="monthly_queue"/>
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="Consumer" method="consume" queue-names="daily_queue" />
</rabbit:listener-container>
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="Consumer" method="consume" queue-names="weekly_queue" />
</rabbit:listener-container>
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="Consumer" method="consume" queue-names="monthly_queue" />
</rabbit:listener-container>
<bean id="Consumer" class="com.test.Consumer" />
</beans>
知道如何用优先级队列解决这个问题?
ps:我也想知道,如果Apache Camel有我可以依赖的东西吗?
更新1:我刚从Apache Camel看到这个:“https://issues.apache.org/jira/browse/CAMEL-2537”JMSPriority上的音序器似乎是我正在寻找的,任何人都曾尝试过这个吗?
更新2:假设我在@Gary Russell推荐下使用RabbitMQ的插件,我有以下spring-rabbitmq上下文XML配置,这似乎有意义(由guest ..):
<rabbit:queue name="ad_google_dfa_reporting_queue">
<rabbit:queue-arguments>
<entry key="x-max-priority" value="10"/>
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="adGoogleDfaReporting" method="consume" queue-names="ad_google_dfa_reporting_queue" />
</rabbit:listener-container>
<bean id="Consumer" class="com.test.Consumer" />
上面的xml配置已成功创建一个名称为“ad_google_dfa_reporting_queue”的队列,并带有参数参数:x-max-priority:10&durable:true
但是,当没有优先发送消息的代码时,我完全失去了它 . 如何在示例URL中定义优先级:https://github.com/rabbitmq/rabbitmq-priority-queue/blob/master/examples/java/src/com/rabbitmq/examples/PriorityQueue.java
AmqpTemplate amqpTemplateGoogleDfaReporting = (AmqpTemplate) applicationContext.getBean("amqpTemplateadGoogleDfaReporting");
amqpTemplateGoogleDfaReporting.convertAndSend("message"); // how to define message priority?
更新3:基于@Gary的答案,我设法发送消息中设置了优先级的消息,如下图所示:
然而,当我发送1000条随机优先级在1-10之间的消息时,消费者正在消费消息各种优先事项 . (我原本只期望首先使用高优先级消息) . 以下是Message producer的代码:
Random random = new Random();
for (int i=0; i< 1000; i++){
final int priority = random.nextInt(10 - 1 + 1) + 1;
DfaReportingModel model = new DfaReportingModel();
model.setReportType(DfaReportingModel.ReportType.FACT);
model.setUserProfileId(0l + priority);
amqpTemplateGoogleDfaReporting.convertAndSend(model, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setPriority(priority);
return message;
}
});
}
以下是Message consumer的代码:
public void consume(DfaReportingModel message) {
System.out.println(message.getUserProfileId());
Thread.sleep(500);
}
得到的结果:
9, 10, 7, 9, 6, 4, 10, 10, 3, 10, 6, 1, 5, 6, 6, 3, 4, 7, 6, 8, 3, 1, 4, 5, 5, 3, 10, 9, 5, 1, 8, 9, 6, 9, 3, 10, 7, 4, 8, 7, 3, 4, 8, 2, 6, 9, 6, 4, 7, 7, 2, 8, 4, 4, 1,
更新4:问题解决了!知道https://github.com/rabbitmq/rabbitmq-priority-queue中的示例代码在我的环境中工作,我认为问题出在Spring环境中 . 因此,经过无数次尝试和错误与不同类型的配置,我指出确切的组合,这将使其工作!如下:
<rabbit:queue name="ad_google_dfa_reporting_queue">
<rabbit:queue-arguments>
<entry key="x-max-priority">
<value type="java.lang.Integer">10</value> <!-- MUST specifically define java.lang.Integer to get it to work -->
</entry>
</rabbit:queue-arguments>
</rabbit:queue>
如果没有专门定义值为Integer类型,则优先级队列不起作用 . 最后,它解决了 . 好极了!
2 回答
RabbitMQ现在有一个priority queue plugin,其中消息按优先级顺序传递 . 使用它而不是重新排列低优先级消息的方案会更好,这些消息在运行时会非常昂贵 .
EDIT:
使用
rabbitTemplate.convertAndSend(...)
方法时,如果要在消息上设置优先级属性,则需要在模板中实现自定义MessagePropertiesConverter
(子类DefaultMessagePropertiesConverter
)或使用后处理消息的convertAnSend
变体;例如 . :从版本3.5.0开始,RabbitMQ在核心中实现了priority queue .
您可以使用
x-max-priority
参数声明优先级队列 . 此参数应为整数,指示队列应支持的最大优先级 . 例如,使用Java客户端:然后,您可以使用
basic.properties
的优先级字段发布优先级消息 . 数字越大表示优先级越高 .