This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new c5bbd674c [ISSUE #3983] Optimize MessageQueue (#3984)
c5bbd674c is described below
commit c5bbd674c4ffbb9f90e869466f9aa4fa0ff7065e
Author: mxsm <[email protected]>
AuthorDate: Wed Jun 7 10:27:38 2023 +0800
[ISSUE #3983] Optimize MessageQueue (#3984)
* [ISSUE #3983] Optimize MessageQueue
* modify public of attribute items to private
* optimize code readability
* optimize code logic
---
.../storage/standalone/broker/MessageQueue.java | 57 +++++++++++++++-------
1 file changed, 40 insertions(+), 17 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
index c887bf693..a7951cb38 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/MessageQueue.java
@@ -24,18 +24,21 @@ import java.util.concurrent.locks.ReentrantLock;
import com.google.common.base.Preconditions;
+import lombok.Getter;
+
/**
* This is a block queue, can get entity by offset. The queue is a FIFO data
structure.
*/
public class MessageQueue {
+ @Getter
private final MessageEntity[] items;
- private int takeIndex;
+ private volatile int takeIndex;
- private int putIndex;
+ private volatile int putIndex;
- private int count;
+ private volatile int count;
private final ReentrantLock lock;
@@ -59,9 +62,10 @@ public class MessageQueue {
}
/**
- * Insert the message at the tail of this queue, waiting for space to
become available if the queue is full
+ * Inserts the specified MessageEntity object into the queue.
*
- * @param messageEntity
+ * @param messageEntity The MessageEntity object to be inserted into the
queue.
+ * @throws InterruptedException if the current thread is interrupted while
waiting for space to become available in the queue.
*/
public void put(MessageEntity messageEntity) throws InterruptedException {
Preconditions.checkNotNull(messageEntity);
@@ -80,8 +84,8 @@ public class MessageQueue {
/**
* Get the first message at this queue, waiting for the message is
available if the queue is empty, this method will not remove the message
*
- * @return MessageEntity
- * @throws InterruptedException
+ * @return The MessageEntity object at the head of the queue.
+ * @throws InterruptedException if the current thread is interrupted while
waiting for an element to become available in the queue.
*/
public MessageEntity take() throws InterruptedException {
ReentrantLock reentrantLock = this.lock;
@@ -145,21 +149,26 @@ public class MessageQueue {
/**
* Get the message by offset, since the offset is increment, so we can get
the first message in this queue and calculate the index of this offset
*
- * @param offset
- * @return MessageEntity
+ * @param offset The offset of the MessageEntity object to be retrieved.
+ * @return The MessageEntity object with the specified offset, or null if
no such object exists in the queue.
+ * @throws RuntimeException if the specified offset is less than the
offset of the head MessageEntity object.
*/
public MessageEntity getByOffset(long offset) {
ReentrantLock reentrantLock = this.lock;
reentrantLock.lock();
try {
- MessageEntity head = getHead();
- if (head == null) {
+ if (count == 0) {
return null;
}
+ int tailIndex = putIndex - 1;
+ MessageEntity head = itemAt(takeIndex);
if (head.getOffset() > offset) {
throw new RuntimeException(String.format("The message has been
deleted, offset: %s", offset));
}
- MessageEntity tail = getTail();
+ if (tailIndex < 0) {
+ tailIndex += items.length;
+ }
+ MessageEntity tail = itemAt(tailIndex);
if (tail == null || tail.getOffset() < offset) {
return null;
}
@@ -174,6 +183,9 @@ public class MessageQueue {
}
}
+ /**
+ * Removes the MessageEntity object at the head of the queue.
+ */
public void removeHead() {
ReentrantLock reentrantLock = this.lock;
reentrantLock.lock();
@@ -195,11 +207,21 @@ public class MessageQueue {
return count;
}
-
+ /**
+ * Returns the MessageEntity object at the specified index.
+ *
+ * @param index The index of the MessageEntity object to be returned.
+ * @return The MessageEntity object at the specified index.
+ */
private MessageEntity itemAt(int index) {
return items[index];
}
+ /**
+ * Insert the message at the tail of this queue, waiting for space to
become available if the queue is full
+ *
+ * @param messageEntity The MessageEntity object to be inserted into the
queue.
+ */
private void enqueue(MessageEntity messageEntity) {
items[putIndex++] = messageEntity;
if (putIndex == items.length) {
@@ -209,6 +231,11 @@ public class MessageQueue {
notEmpty.signalAll();
}
+ /**
+ * Removes and returns the MessageEntity object at the head of the queue.
+ *
+ * @return The MessageEntity object at the head of the queue.
+ */
private MessageEntity dequeue() {
MessageEntity item = items[takeIndex++];
if (takeIndex == items.length) {
@@ -225,8 +252,4 @@ public class MessageQueue {
public int getPutIndex() {
return putIndex;
}
-
- public MessageEntity[] getItems() {
- return items;
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
