github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

PriorityBlockingDeque implemented. #1235

Browse Source
pull/1249/head
Nikita 7 years ago
parent c9dacc4007
commit 626f5785f6
6 changed files with 463 additions and 11 deletions
Show Stats Download Patch File Download Diff File

12
redisson/src/main/java/org/redisson/Redisson.java
Unescape Escape View File

@ -54,6 +54,7 @@ import org.redisson.api.RMap;
import org.redisson.api.RMapCache;
import org.redisson.api.RPatternTopic;
import org.redisson.api.RPermitExpirableSemaphore;
import org.redisson.api.RPriorityBlockingDeque;
import org.redisson.api.RPriorityBlockingQueue;
import org.redisson.api.RPriorityDeque;
import org.redisson.api.RPriorityQueue;
@ -654,6 +655,17 @@ public class Redisson implements RedissonClient {
public <V> RPriorityBlockingQueue<V> getPriorityBlockingQueue(String name, Codec codec) {
return new RedissonPriorityBlockingQueue<V>(codec, connectionManager.getCommandExecutor(), name, this);
}
@Override
public <V> RPriorityBlockingDeque<V> getPriorityBlockingDeque(String name) {
return new RedissonPriorityBlockingDeque<V>(connectionManager.getCommandExecutor(), name, this);
}
@Override
public <V> RPriorityBlockingDeque<V> getPriorityBlockingDeque(String name, Codec codec) {
return new RedissonPriorityBlockingDeque<V>(codec, connectionManager.getCommandExecutor(), name, this);
}
@Override
public <V> RPriorityDeque<V> getPriorityDeque(String name) {

240
redisson/src/main/java/org/redisson/RedissonPriorityBlockingDeque.java
Unescape Escape View File

@ -0,0 +1,240 @@
/**
* Copyright 2016 Nikita Koksharov
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.redisson;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import org.redisson.api.RFuture;
import org.redisson.api.RPriorityBlockingDeque;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.command.CommandExecutor;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise;
/**
* <p>Distributed and concurrent implementation of priority blocking deque.
*
* <p>Queue size limited by Redis server memory amount. This is why {@link #remainingCapacity()} always
* returns <code>Integer.MAX_VALUE</code>
*
* @author Nikita Koksharov
*/
public class RedissonPriorityBlockingDeque<V> extends RedissonPriorityDeque<V> implements RPriorityBlockingDeque<V> {
private final RedissonPriorityBlockingQueue<V> blockingQueue;
protected RedissonPriorityBlockingDeque(CommandExecutor commandExecutor, String name, RedissonClient redisson) {
super(commandExecutor, name, redisson);
blockingQueue = (RedissonPriorityBlockingQueue<V>) redisson.getPriorityBlockingQueue(name);
}
protected RedissonPriorityBlockingDeque(Codec codec, CommandExecutor commandExecutor, String name, RedissonClient redisson) {
super(codec, commandExecutor, name, redisson);
blockingQueue = (RedissonPriorityBlockingQueue<V>) redisson.getPriorityBlockingQueue(name, codec);
}
@Override
public void put(V e) throws InterruptedException {
add(e);
}
@Override
public boolean offer(V e, long timeout, TimeUnit unit) throws InterruptedException {
return offer(e);
}
@Override
public RFuture<V> takeAsync() {
return blockingQueue.takeAsync();
}
@Override
public V take() throws InterruptedException {
return blockingQueue.take();
}
public RFuture<V> pollAsync(long timeout, TimeUnit unit) {
return blockingQueue.pollAsync(timeout, unit);
}
@Override
public V poll(long timeout, TimeUnit unit) throws InterruptedException {
return blockingQueue.poll(timeout, unit);
}
@Override
public V pollFromAny(long timeout, TimeUnit unit, String ... queueNames) throws InterruptedException {
throw new UnsupportedOperationException("use poll method");
}
@Override
public RFuture<V> pollLastAndOfferFirstToAsync(String queueName, long timeout, TimeUnit unit) {
return blockingQueue.pollLastAndOfferFirstToAsync(queueName, timeout, unit);
}
@Override
public V pollLastAndOfferFirstTo(String queueName, long timeout, TimeUnit unit) throws InterruptedException {
return blockingQueue.pollLastAndOfferFirstTo(queueName, timeout, unit);
}
@Override
public V takeLastAndOfferFirstTo(String queueName) throws InterruptedException {
return get(takeLastAndOfferFirstToAsync(queueName));
}
public RFuture<V> takeLastAndOfferFirstToAsync(String queueName) {
return pollLastAndOfferFirstToAsync(queueName, 0, TimeUnit.SECONDS);
}
@Override
public int remainingCapacity() {
return Integer.MAX_VALUE;
}
@Override
public int drainTo(Collection<? super V> c) {
return blockingQueue.drainTo(c);
}
public RFuture<Integer> drainToAsync(Collection<? super V> c) {
return blockingQueue.drainToAsync(c);
}
@Override
public int drainTo(Collection<? super V> c, int maxElements) {
return blockingQueue.drainTo(c, maxElements);
}
public RFuture<Integer> drainToAsync(Collection<? super V> c, int maxElements) {
return blockingQueue.drainToAsync(c, maxElements);
}
@Override
public RFuture<Boolean> offerAsync(V e) {
throw new UnsupportedOperationException("use offer method");
}
@Override
public RFuture<V> pollFromAnyAsync(long timeout, TimeUnit unit, String... queueNames) {
throw new UnsupportedOperationException("use poll method");
}
@Override
public RFuture<Void> putAsync(V e) {
throw new UnsupportedOperationException("use add method");
}
@Override
public RFuture<Void> putFirstAsync(V e) {
return addFirstAsync(e);
}
@Override
public RFuture<Void> putLastAsync(V e) {
return addLastAsync(e);
}
@Override
public void putFirst(V e) throws InterruptedException {
addFirst(e);
}
@Override
public void putLast(V e) throws InterruptedException {
addLast(e);
}
@Override
public boolean offerFirst(V e, long timeout, TimeUnit unit) throws InterruptedException {
addFirst(e);
return true;
}
@Override
public boolean offerLast(V e, long timeout, TimeUnit unit) throws InterruptedException {
addLast(e);
return true;
}
@Override
public V takeFirst() throws InterruptedException {
return get(takeFirstAsync());
}
@Override
public RFuture<V> takeFirstAsync() {
return takeAsync();
}
@Override
public RFuture<V> takeLastAsync() {
RPromise<V> result = new RedissonPromise<V>();
blockingQueue.takeAsync(result, 0, 0, RedisCommands.RPOP, getName());
return result;
}
@Override
public V takeLast() throws InterruptedException {
return get(takeLastAsync());
}
@Override
public RFuture<V> pollFirstAsync(long timeout, TimeUnit unit) {
return pollAsync(timeout, unit);
}
@Override
public V pollFirstFromAny(long timeout, TimeUnit unit, String ... queueNames) throws InterruptedException {
return get(pollFirstFromAnyAsync(timeout, unit, queueNames));
}
@Override
public RFuture<V> pollFirstFromAnyAsync(long timeout, TimeUnit unit, String ... queueNames) {
return pollFromAnyAsync(timeout, unit, queueNames);
}
@Override
public V pollLastFromAny(long timeout, TimeUnit unit, String ... queueNames) throws InterruptedException {
return get(pollLastFromAnyAsync(timeout, unit, queueNames));
}
@Override
public RFuture<V> pollLastFromAnyAsync(long timeout, TimeUnit unit, String ... queueNames) {
throw new UnsupportedOperationException();
}
@Override
public V pollFirst(long timeout, TimeUnit unit) throws InterruptedException {
return get(pollFirstAsync(timeout, unit));
}
@Override
public RFuture<V> pollLastAsync(long timeout, TimeUnit unit) {
RPromise<V> result = new RedissonPromise<V>();
blockingQueue.takeAsync(result, 0, unit.toMicros(timeout), RedisCommands.RPOP, getName());
return result;
}
@Override
public V pollLast(long timeout, TimeUnit unit) throws InterruptedException {
return get(pollLastAsync(timeout, unit));
}
}

47
redisson/src/main/java/org/redisson/RedissonPriorityDeque.java
Unescape Escape View File

@ -20,6 +20,7 @@ import java.util.NoSuchElementException;
import org.redisson.api.RFuture;
import org.redisson.api.RPriorityDeque;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.RedisCommands;
@ -38,22 +39,30 @@ public class RedissonPriorityDeque<V> extends RedissonPriorityQueue<V> implement
private static final RedisCommand<Object> LRANGE_SINGLE = new RedisCommand<Object>("LRANGE", new ListFirstObjectDecoder());
protected RedissonPriorityDeque(CommandExecutor commandExecutor, String name, Redisson redisson) {
protected RedissonPriorityDeque(CommandExecutor commandExecutor, String name, RedissonClient redisson) {
super(commandExecutor, name, redisson);
}
public RedissonPriorityDeque(Codec codec, CommandExecutor commandExecutor, String name, Redisson redisson) {
public RedissonPriorityDeque(Codec codec, CommandExecutor commandExecutor, String name, RedissonClient redisson) {
super(codec, commandExecutor, name, redisson);
}
public RFuture<Void> addFirstAsync(V e) {
throw new UnsupportedOperationException("use add or put method");
}
public RFuture<Void> addLastAsync(V e) {
throw new UnsupportedOperationException("use add or put method");
}
@Override
public void addFirst(V e) {
throw new UnsupportedOperationException();
throw new UnsupportedOperationException("use add or put method");
}
@Override
public void addLast(V e) {
throw new UnsupportedOperationException();
throw new UnsupportedOperationException("use add or put method");
}
@Override
@ -108,14 +117,22 @@ public class RedissonPriorityDeque<V> extends RedissonPriorityQueue<V> implement
@Override
public boolean offerFirst(V e) {
throw new UnsupportedOperationException();
throw new UnsupportedOperationException("use add or put method");
}
public RFuture<Boolean> offerFirstAsync(V e) {
throw new UnsupportedOperationException("use add or put method");
}
@Override
public boolean offerLast(V e) {
throw new UnsupportedOperationException();
throw new UnsupportedOperationException("use add or put method");
}
public RFuture<Boolean> offerLastAsync(V e) {
throw new UnsupportedOperationException("use add or put method");
}
// @Override
public RFuture<V> peekFirstAsync() {
return getAsync(0);
@ -128,12 +145,20 @@ public class RedissonPriorityDeque<V> extends RedissonPriorityQueue<V> implement
@Override
public V peekLast() {
return get(getLastAsync());
return get(peekLastAsync());
}
public RFuture<V> peekLastAsync() {
return getLastAsync();
}
@Override
public V pollFirst() {
return poll();
return get(pollFirstAsync());
}
public RFuture<V> pollFirstAsync() {
return pollAsync();
}
public RFuture<V> pollLastAsync() {
@ -157,9 +182,13 @@ public class RedissonPriorityDeque<V> extends RedissonPriorityQueue<V> implement
@Override
public void push(V e) {
throw new UnsupportedOperationException();
throw new UnsupportedOperationException("use add or put method");
}
public RFuture<Void> pushAsync(V e) {
throw new UnsupportedOperationException("use add or put method");
}
// @Override
public RFuture<Boolean> removeFirstOccurrenceAsync(Object o) {
return removeAsync(o, 1);

26
redisson/src/main/java/org/redisson/api/RPriorityBlockingDeque.java
Unescape Escape View File

@ -0,0 +1,26 @@
/**
* Copyright 2016 Nikita Koksharov
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.redisson.api;
/**
* RPriorityBlockingDeque backed by Redis
*
* @author Nikita Koksharov
* @param <V> the type of elements held in this collection
*/
public interface RPriorityBlockingDeque<V> extends RBlockingDeque<V>, RPriorityDeque<V> {
}

25
redisson/src/main/java/org/redisson/api/RedissonClient.java
Unescape Escape View File

@ -612,7 +612,7 @@ public interface RedissonClient {
<V> RPriorityQueue<V> getPriorityQueue(String name, Codec codec);
/**
* Returns priority unbounded blocking queue instance by name.
* Returns unbounded priority blocking queue instance by name.
* It uses comparator to sort objects.
*
* @param <V> type of value
@ -622,7 +622,7 @@ public interface RedissonClient {
<V> RPriorityBlockingQueue<V> getPriorityBlockingQueue(String name);
/**
* Returns priority unbounded blocking queue instance by name
* Returns unbounded priority blocking queue instance by name
* using provided codec for queue objects.
* It uses comparator to sort objects.
*
@ -633,6 +633,27 @@ public interface RedissonClient {
*/
<V> RPriorityBlockingQueue<V> getPriorityBlockingQueue(String name, Codec codec);
/**
* Returns unbounded priority blocking deque instance by name.
* It uses comparator to sort objects.
*
* @param <V> type of value
* @param name of object
* @return Queue object
*/
<V> RPriorityBlockingDeque<V> getPriorityBlockingDeque(String name);
/**
* Returns unbounded priority blocking deque instance by name
* using provided codec for queue objects.
* It uses comparator to sort objects.
*
* @param <V> type of value
* @param name - name of object
* @param codec - codec for message
* @return Queue object
*/
<V> RPriorityBlockingDeque<V> getPriorityBlockingDeque(String name, Codec codec);
/**
* Returns priority unbounded deque instance by name.

124
redisson/src/test/java/org/redisson/RedissonPriorityBlockingDequeTest.java
Unescape Escape View File

@ -0,0 +1,124 @@
package org.redisson;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.redisson.api.RBlockingDeque;
public class RedissonPriorityBlockingDequeTest extends BaseTest {
@Test(timeout = 3000)
public void testShortPoll() throws InterruptedException {
RBlockingDeque<Integer> queue = redisson.getPriorityBlockingDeque("queue:pollany");
queue.pollLastAsync(500, TimeUnit.MILLISECONDS);
queue.pollFirstAsync(10, TimeUnit.MICROSECONDS);
}
@Test
public void testTakeFirst() throws InterruptedException {
RBlockingDeque<Integer> deque = redisson.getPriorityBlockingDeque("queue:take");
deque.add(1);
deque.add(2);
deque.add(3);
deque.add(4);
assertThat(deque.takeFirst()).isEqualTo(1);
assertThat(deque.takeFirst()).isEqualTo(2);
assertThat(deque.takeFirst()).isEqualTo(3);
assertThat(deque.takeFirst()).isEqualTo(4);
assertThat(deque.size()).isZero();
}
@Test
public void testTakeLast() throws InterruptedException {
RBlockingDeque<Integer> deque = redisson.getPriorityBlockingDeque("queue:take");
deque.add(1);
deque.add(2);
deque.add(3);
deque.add(4);
assertThat(deque.takeLast()).isEqualTo(4);
assertThat(deque.takeLast()).isEqualTo(3);
assertThat(deque.takeLast()).isEqualTo(2);
assertThat(deque.takeLast()).isEqualTo(1);
assertThat(deque.size()).isZero();
}
@Test
public void testTakeFirstAwait() throws InterruptedException {
RBlockingDeque<Integer> deque = redisson.getPriorityBlockingDeque("queue:take");
Executors.newSingleThreadScheduledExecutor().schedule(() -> {
RBlockingDeque<Integer> deque1 = redisson.getBlockingDeque("queue:take");
deque1.add(1);
deque1.add(2);
deque1.add(3);
deque1.add(4);
}, 10, TimeUnit.SECONDS);
long s = System.currentTimeMillis();
assertThat(deque.takeFirst()).isEqualTo(1);
assertThat(System.currentTimeMillis() - s).isGreaterThan(9000);
Thread.sleep(50);
assertThat(deque.takeFirst()).isEqualTo(2);
assertThat(deque.takeFirst()).isEqualTo(3);
assertThat(deque.takeFirst()).isEqualTo(4);
}
@Test
public void testTakeLastAwait() throws InterruptedException {
RBlockingDeque<Integer> deque = redisson.getPriorityBlockingDeque("queue:take");
Executors.newSingleThreadScheduledExecutor().schedule(() -> {
RBlockingDeque<Integer> deque1 = redisson.getBlockingDeque("queue:take");
deque1.add(1);
deque1.add(2);
deque1.add(3);
deque1.add(4);
}, 10, TimeUnit.SECONDS);
long s = System.currentTimeMillis();
assertThat(deque.takeLast()).isEqualTo(4);
assertThat(System.currentTimeMillis() - s).isGreaterThan(9000);
Thread.sleep(50);
assertThat(deque.takeLast()).isEqualTo(3);
assertThat(deque.takeLast()).isEqualTo(2);
assertThat(deque.takeLast()).isEqualTo(1);
}
@Test
public void testPollFirst() throws InterruptedException {
RBlockingDeque<Integer> queue1 = redisson.getPriorityBlockingDeque("queue1");
queue1.put(1);
queue1.put(2);
queue1.put(3);
assertThat(queue1.pollFirst(2, TimeUnit.SECONDS)).isEqualTo(1);
assertThat(queue1.pollFirst(2, TimeUnit.SECONDS)).isEqualTo(2);
assertThat(queue1.pollFirst(2, TimeUnit.SECONDS)).isEqualTo(3);
long s = System.currentTimeMillis();
assertThat(queue1.pollFirst(5, TimeUnit.SECONDS)).isNull();
assertThat(System.currentTimeMillis() - s).isGreaterThan(5000);
}
@Test
public void testPollLast() throws InterruptedException {
RBlockingDeque<Integer> queue1 = redisson.getPriorityBlockingDeque("queue1");
queue1.add(3);
queue1.add(1);
queue1.add(2);
assertThat(queue1.pollLast(2, TimeUnit.SECONDS)).isEqualTo(3);
assertThat(queue1.pollLast(2, TimeUnit.SECONDS)).isEqualTo(2);
assertThat(queue1.pollLast(2, TimeUnit.SECONDS)).isEqualTo(1);
long s = System.currentTimeMillis();
assertThat(queue1.pollLast(5, TimeUnit.SECONDS)).isNull();
assertThat(System.currentTimeMillis() - s).isGreaterThanOrEqualTo(5000);
}
}
Write Preview
Loading…
Cancel
Save