Spring TaskExecutor 实现队列优先级
Spring TaskExecutor Implementation Queue Priorization
我正在处理的应用程序收到来自外部系统的通知,我想处理这些通知。
到目前为止我有以下实现:
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 个线程,因为由于一些难以更改的遗留问题,我不想并行执行通知)
<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)
}
});
我拥有的这个通知对象包含一个时间戳字段。
我想通过此字段对队列中的通知进行优先级排序(我认为 Spring 默认使用无界队列,这对我来说更好,因为我需要一个无界队列)
我能否以某种方式将其集成到我的 spring 应用程序中,而无需从头开始手动实施?所以我想根据通知对象上的时间戳字段对队列中的任务(可运行对象)进行排序。(这是我传递给 "processNotification" 方法的对象)
ThreadPoolTaskExecutor
得到 BlockingQueue
:
的支持
protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
if (queueCapacity > 0) {
return new LinkedBlockingQueue<Runnable>(queueCapacity);
}
else {
return new SynchronousQueue<Runnable>();
}
}
如果你想对你的任务进行排序,你需要覆盖这个函数以允许优先排序:
public class YourPool extends ThreadPoolTaskExecutor {
@Override
protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
return new PriorityBlockingQueue<>(queueCapacity);
}
}
您提交的任务必须具有可比性:
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().compareTo(other.notification.getTimestamp());
}
}
然后提交你的任务:
asynchronousService.executeAsynchronously(new YourTask(notificationX));
我正在处理的应用程序收到来自外部系统的通知,我想处理这些通知。
到目前为止我有以下实现:
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 个线程,因为由于一些难以更改的遗留问题,我不想并行执行通知)
<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)
}
});
我拥有的这个通知对象包含一个时间戳字段。 我想通过此字段对队列中的通知进行优先级排序(我认为 Spring 默认使用无界队列,这对我来说更好,因为我需要一个无界队列)
我能否以某种方式将其集成到我的 spring 应用程序中,而无需从头开始手动实施?所以我想根据通知对象上的时间戳字段对队列中的任务(可运行对象)进行排序。(这是我传递给 "processNotification" 方法的对象)
ThreadPoolTaskExecutor
得到 BlockingQueue
:
protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
if (queueCapacity > 0) {
return new LinkedBlockingQueue<Runnable>(queueCapacity);
}
else {
return new SynchronousQueue<Runnable>();
}
}
如果你想对你的任务进行排序,你需要覆盖这个函数以允许优先排序:
public class YourPool extends ThreadPoolTaskExecutor {
@Override
protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
return new PriorityBlockingQueue<>(queueCapacity);
}
}
您提交的任务必须具有可比性:
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().compareTo(other.notification.getTimestamp());
}
}
然后提交你的任务:
asynchronousService.executeAsynchronously(new YourTask(notificationX));