Spring TaskExecutor 实现队列优先级

本文介绍了Spring TaskExecutor 实现队列优先级的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在处理的应用程序接收来自我想要处理的外部系统的通知.

The application I am working on receives notifications from external systems, which I want to process.

到目前为止，我有以下实现:

Till now I have the following implementation:

public class AsynchronousServiceImpl implements AsynchronousService { private TaskExecutor taskExecutor; @Override public void executeAsynchronously(Runnable task) { taskExecutor.execute(task); } @Required public void setTaskExecutor(TaskExecutor taskExecutor) { this.taskExecutor = taskExecutor; } }

spring 配置(我只需要 1 个线程，因为由于一些难以更改的遗留问题，我不想并行执行通知)

spring configuration (I only need 1 thread since I don't want to execute the notifications in parallel due some legacy issues which are hard to change)

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <property name="corePoolSize" value="1"/> <property name="maxPoolSize" value="1"/> <property name="WaitForTasksToCompleteOnShutdown" value="true"/> </bean>

我在这里执行代码:

asynchronousService.executeAsynchronously(new Runnable() { @Override public void run() { someMethod.processNotification(notification) } });

我拥有的这个 Notification 对象包含一个时间戳字段.我想通过这个字段对队列中的通知进行优先级排序(我认为 Spring 默认使用一个无界队列，这对我来说更好，因为我需要一个无界队列)

This Notification object which I have contains a timestamp field. I want to prioritize the notifications in the queue by this field (I think Spring uses as default an unbounded queue which is fine more me, since I need an unbounded queue)

我可以在不从头开始手动实现的情况下以某种方式将其集成到我的 spring 应用程序中吗?所以我想根据通知对象上的时间戳字段对队列中的任务(可运行对象)进行排序.(这是我传递给processNotification"方法的对象)

Can I integrate this somehow in my spring application without implementing it manually from scratch? So I want to sort the taskss (runnable-objects) in the queue based on the timestamp field on the notification object.(It is that object that I am passing to the "processNotification" method)

推荐答案

ThreadPoolTask​​Executor 由 BlockingQueue 支持:

protected BlockingQueue<Runnable> createQueue(int queueCapacity) { if (queueCapacity > 0) { return new LinkedBlockingQueue<Runnable>(queueCapacity); } else { return new SynchronousQueue<Runnable>(); } }

如果要对任务进行排序，则需要覆盖此函数以允许优先排序:

If you want to order your task, you need to override this function to allow priority ordering:

public class YourPool extends ThreadPoolTaskExecutor { @Override protected BlockingQueue<Runnable> createQueue(int queueCapacity) { return new PriorityBlockingQueue<>(queueCapacity); } }

您提交的任务必须具有可比性:

Your submitted task must be comparable:

public class YourTask implements Runnable, Comparable<YourTask> { private Notification notification; public YourTask(Notification notification) { this.notification = notification; } @Override public void run() { someMethod.processNotification(notification) } @Override public int compareTo(B other) { // Here you implement the priority return notification.getTimestamp()pareTo(other.notification.getTimestamp()); } }

然后提交您的任务:

asynchronousService.executeAsynchronously(new YourTask(notificationX));