Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Optimise the postgres queries for popping messages #76

Merged
merged 1 commit into from
Feb 21, 2024
Merged

Optimise the postgres queries for popping messages #76

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 9 additions & 35 deletions postgres-persistence/src/main/java/com/netflix/conductor/postgres/dao/PostgresQueueDAO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -422,17 +422,19 @@ private boolean removeMessage(Connection connection, String queueName, String me
q -> q.addParameter(queueName).addParameter(messageId).executeDelete());
}

private List<Message> peekMessages(Connection connection, String queueName, int count) {
if (count < 1) {
return Collections.emptyList();
}
private List<Message> popMessages(
Connection connection, String queueName, int count, int timeout) {

final String PEEK_MESSAGES =
"SELECT message_id, priority, payload FROM queue_message WHERE queue_name = ? AND popped = false AND deliver_on <= (current_timestamp + (1000 ||' microseconds')::interval) ORDER BY priority DESC, deliver_on, created_on LIMIT ? FOR UPDATE SKIP LOCKED";
String POP_QUERY =
"UPDATE queue_message SET popped = true WHERE message_id IN ("
+ "SELECT message_id FROM queue_message WHERE queue_name = ? AND popped = false AND "
+ "deliver_on <= (current_timestamp + (1000 ||' microseconds')::interval) "
+ "ORDER BY priority DESC, deliver_on, created_on LIMIT ? FOR UPDATE SKIP LOCKED"
+ ") RETURNING message_id, priority, payload";

return query(
connection,
PEEK_MESSAGES,
POP_QUERY,
p ->
p.addParameter(queueName)
.addParameter(count)
Expand All @@ -450,34 +452,6 @@ private List<Message> peekMessages(Connection connection, String queueName, int
}));
}

private List<Message> popMessages(
Connection connection, String queueName, int count, int timeout) {
List<Message> messages = peekMessages(connection, queueName, count);

if (messages.isEmpty()) {
return messages;
}

List<Message> poppedMessages = new ArrayList<>();
for (Message message : messages) {
final String POP_MESSAGE =
"UPDATE queue_message SET popped = true WHERE queue_name = ? AND message_id = ? AND popped = false";
int result =
query(
connection,
POP_MESSAGE,
q ->
q.addParameter(queueName)
.addParameter(message.getId())
.executeUpdate());

if (result == 1) {
poppedMessages.add(message);
}
}
return poppedMessages;
}

@Override
public boolean containsMessage(String queueName, String messageId) {
return getWithRetriedTransactions(tx -> existsMessage(tx, queueName, messageId));
Expand Down
Loading