Как создать канал очереди LIFO

Question

В настоящее время я использую канал очереди в своем потоке интеграции, но он использует способ извлечения FIFO. Есть ли способ изменить его на LIFO?

Кроме того, есть ли способ удалить сообщения из очереди на основе свойства?

Я подозреваю, что мне нужно будет использовать PriorityChannel для переключения с FIFO на LIFO, но я не могу понять, как это будет сделано.

@Bean
    fun notificationChannel(): MessageChannel {
        return MessageChannels.queue().get()
    }

Например, мой QueueChannel будет заполнен сообщениями, содержащими userId. Поскольку меня интересует только последнее состояние сообщения userId, я хочу использовать LIFO и удалить все сообщения с тем же userId, что и у последнего сообщения.

Answer 1

Канал очереди принимает очередь в качестве параметра конструктора public QueueChannel(Queue<Message<?>> queue), а DSL с пружинной интеграцией предоставляет public static QueueChannelSpec queue(Queue<Message<?>> queue). Таким образом, вы можете использовать Collections.asLifoQueue(..) в качестве аргумента для вышеуказанного фабричного метода, чтобы получить желаемое поведение.

Answer 2

Звучит так, будто вы на самом деле не хотите LIFO, вам просто нужна последняя версия для данного условия.

Но, независимо от того, Queue - довольно простой интерфейс; он имеет только несколько методов, а QueueChannel использует только poll(), offer() и size().

Так что должно быть просто создать собственную очередь, например, основанный на потоке-безопасном ConcurrentHashMap<String, Message<?>> с ключом, являющимся вашим состоянием.