java - 如果有更重要的消息进来，则添加到消息队列

Question

我们需要首先发送具有最高优先级的消息，因此我们使用 PriorityQueue 来达到我们的目的。

PriorityQueue<MessageData> queue = new PriorityQueue<MessageData>();

但是，我们也希望我们的队列表现得像一个排序集。因此，我们调整 PriorityQueue 以忽略重复现有成员的插入。

import java.util.Comparator;
import java.util.PriorityQueue;

public class PrioritySet<E> extends PriorityQueue<E> {

    private static final long serialVersionUID = 34658778L;

    public PrioritySet() {
        super();
    }

    public PrioritySet(int initialCapacity, Comparator<? super E> comparator) {
        super(initialCapacity, comparator);
    }

    @Override
    public boolean offer(E e) {
        boolean isAdded = false;
        if(!super.contains(e)) {
            isAdded = super.offer(e);
        }
        return isAdded;
    }
}

现在我们的应用程序具体实现了数据结构。

import java.util.Comparator;

public class MessagePrioritySet extends PrioritySet<MessageData> {

    private static final long serialVersionUID = 34658779L;

    private int minPriorityNumber;

    public MessagePrioritySet() {
        super();
    }

    public MessagePrioritySet(int initialCapacity, Comparator<MessageData> comparator) {
        super(initialCapacity, comparator);
    }

    public synchronized int getMinPriorityNumber() {
        return minPriorityNumber;
    }

    public synchronized void setMinPriorityNumber(int minPriorityNumber) {
        this.minPriorityNumber = minPriorityNumber;
    }

    @Override
    public synchronized boolean offer(MessageData notification) {
        boolean isAdded = super.offer(notification);
        if (notification.getPriority() < minPriorityNumber)
            minPriorityNumber = notification.getPriority();
        return isAdded;
    }

    public synchronized void reportSent(MessageData notification) {
        MessageData nextMessageData = peek();
        if (nextMessageData == null)
            minPriorityNumber = 0;
        else if (nextMessageData.getPriority() > notification.getPriority())
            minPriorityNumber = nextMessageData.getPriority();
    }
}

在这里，我们希望数据结构了解消息的最小优先级值，因此我们为此声明一个实例变量。检查传入消息的优先级，如果该优先级低于存储的值，则更新存储的值。需要使用该类来报告任何已发送的消息。如果数据结构中没有其他成员的优先级与被删除的成员一样低，则下一个元素的优先级将成为存储的优先级。

两个线程共享已实现的队列。一个线程从数据库中获取数据并将其插入队列中。另一个读取队列并发送具有最低优先级编号的最高优先级消息。因为队列将最小优先级值设置为0，并且如果存储的最小值不为零，则从数据库获取数据的线程将读取优先级值低于或等于队列中存储的最小值的行，所以我们可以非常确定当队列中的当前消息正在发送时，只有比队列中已有的消息更重要的新消息才会被添加到队列中。

我们认为线程中 while 循环中的操作应该是原子的，并且感谢任何能够告诉我们如何使它们原子化的人。

private void startMptSender() {
    sleepInterval = 1000;
    final MessagePrioritySet messagePrioritySet = new MessagePrioritySet();

    Runnable mptReader = new Runnable() {

        @Override
        public void run() {
            while (true) {
                List<MessageData> messageDataList;

                if (messagePrioritySet.getMinPriorityNumber() == 0)
                    messageDataList = messageDao.readSMSMpt();
                else
                    messageDataList = messageDao.readSMSMpt(messagePrioritySet.getMinPriorityNumber());

                for (MessageData messageData : messageDataList) {
                    messagePrioritySet.offer(messageData);
                }
                try {
                    Thread.sleep(sleepInterval);
                } catch (InterruptedException ie) {

                }
            }
        }
    };

    executor.execute(mptReader);

    Runnable mptPusher = new Runnable() {

        @Override
        public void run() {
            while (status) {
                if (messagePrioritySet.size() > 0) {

                    while (messagePrioritySet.size() != 0) {
                        MessageData noti = messagePrioritySet.remove();
                        mptSender.sendSms(noti);
                        messageDao.markNotificationAsRead(noti.getSyskey());
                        messagePrioritySet.reportSent(noti);
                        try {
                            Thread.sleep(sleepInterval);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                } else {
                    try {
                        Thread.sleep(sleepInterval);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    executor.execute(mptPusher);
}

}

Answer 1

我假设您对原子的意思翻译为:您希望每个线程在一次迭代中完成其所有工作而不被另一个线程中断。

换句话说:你有(可能有多个)操作；当线程 A 正在执行其操作时，线程 B 不应该执行任何操作 - 因为您想确保 B 只能看到 A 所做的更新的“完整集”。

当然，例如，当该操作只是写入一个 int 时，您可以使用 AtomicInteger。但是当您谈论多个操作时......您需要其他东西。

“强力”解决方案是添加某种锁定。含义:您的线程共享一些 LOCK 对象；每当一个线程进入“临界区”时......它需要首先获取该锁(当然之后直接释放)。但这需要非常仔细的设计；因为想要确保线程 A 不会因为持有该锁太长时间而“挨饿”B。

再仔细看看你的代码......也许你可以尝试让你的 minPriority 成为一个 AtomicInteger；问题是这与正在处理队列“大小”的其他线程有何关系。