Java是否存在优先级相同时按FIFO运行的PriorityBlockingQueue实现？

Java标准库中没有直接提供这种「优先级相同时保持FIFO顺序」的PriorityBlockingQueue实现。JDK自带的PriorityBlockingQueue基于二叉堆实现，当多个元素优先级相同时，它们的出队顺序是不确定的，完全依赖堆的内部调整逻辑，无法保证插入顺序。

解决思路：自定义有序优先级元素 #

要实现需求，核心是让同优先级的元素能按插入顺序排序。可以给每个命令元素添加一个全局递增的序列号（或插入时间戳），在元素的比较逻辑中：

• 先按命令优先级比较（比如FLUSH_STREAMS优先级最高）
• 优先级相同时，按序列号的大小比较（序列号小的先插入，先出队）

代码示例 #

首先定义命令的基础类，包含优先级和序列号：

import java.util.concurrent.atomic.AtomicLong;

public abstract class QueueCommand implements Comparable<QueueCommand> {
    // 全局递增序列号，保证插入顺序
    private static final AtomicLong SEQUENCE_GENERATOR = new AtomicLong(0);
    private final long sequence;
    private final int priority;

    protected QueueCommand(int priority) {
        this.priority = priority;
        this.sequence = SEQUENCE_GENERATOR.getAndIncrement();
    }

    @Override
    public int compareTo(QueueCommand other) {
        // 先比较优先级，优先级高的排前面
        if (this.priority != other.priority) {
            return Integer.compare(other.priority, this.priority);
        }
        // 优先级相同则按序列号（插入顺序）排序
        return Long.compare(this.sequence, other.sequence);
    }

    // 子类实现具体命令逻辑，比如AudioFrameCommand、FlushStreamsCommand等
}

然后定义具体的命令子类：

// 音频帧命令，优先级设为1（较低）
public class AudioFrameCommand extends QueueCommand {
    private final byte[] frameData;

    public AudioFrameCommand(byte[] frameData) {
        super(1);
        this.frameData = frameData;
    }

    // 业务方法
    public byte[] getFrameData() {
        return frameData;
    }
}

// FLUSH命令，优先级设为10（最高）
public class FlushStreamsCommand extends QueueCommand {
    public FlushStreamsCommand() {
        super(10);
    }
}

// START命令，优先级可设为10（和FLUSH同级别，按插入顺序处理）
public class StartStreamsCommand extends QueueCommand {
    public StartStreamsCommand() {
        super(10);
    }
}

最后使用PriorityBlockingQueue<QueueCommand>即可：

import java.util.concurrent.PriorityBlockingQueue;

public class CommandQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        PriorityBlockingQueue<QueueCommand> queue = new PriorityBlockingQueue<>();
        
        // 插入几个音频帧
        queue.add(new AudioFrameCommand(new byte[]{1,2,3}));
        queue.add(new AudioFrameCommand(new byte[]{4,5,6}));
        // 插入FLUSH命令
        queue.add(new FlushStreamsCommand());
        // 插入另一个音频帧
        queue.add(new AudioFrameCommand(new byte[]{7,8,9}));
        
        // 消费队列时，FLUSH会先被取出，同优先级的音频帧按插入顺序处理
        while (!queue.isEmpty()) {
            QueueCommand cmd = queue.take();
            // 处理命令逻辑
        }
    }
}

额外说明 #

这种方式既满足了FLUSH_STREAMS命令的高优先级需求，又保证了同优先级的音频帧严格按FIFO顺序处理，完全适配你的业务场景。

内容的提问来源于stack exchange，提问作者user2741831