MapWriter (write-behind) support for RMap added. #927
Nikita
8 years ago
parent
eae170f3a7
commit
c3ef66b23d
@ -0,0 +1,97 @@
|
||||
/**
|
||||
* 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.Queue;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.redisson.command.CommandAsyncExecutor;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import io.netty.util.concurrent.Future;
|
||||
import io.netty.util.concurrent.FutureListener;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <R> return type
|
||||
*/
|
||||
public class MapWriteBehindListener<R> implements FutureListener<R> {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(MapWriteBehindListener.class);
|
||||
|
||||
private static final AtomicBoolean sent = new AtomicBoolean();
|
||||
private static final Queue<Runnable> operations = new ConcurrentLinkedQueue<Runnable>();
|
||||
|
||||
private final MapWriterTask<R> task;
|
||||
private final CommandAsyncExecutor commandExecutor;
|
||||
|
||||
public MapWriteBehindListener(CommandAsyncExecutor commandExecutor, MapWriterTask<R> task) {
|
||||
super();
|
||||
this.commandExecutor = commandExecutor;
|
||||
this.task = task;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void operationComplete(Future<R> future) throws Exception {
|
||||
if (future.isSuccess() && task.condition(future)) {
|
||||
enqueueRunnable(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
task.execute();
|
||||
} catch (Exception e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private void enqueueRunnable(Runnable runnable) {
|
||||
if (runnable != null) {
|
||||
operations.add(runnable);
|
||||
}
|
||||
|
||||
if (sent.compareAndSet(false, true)) {
|
||||
commandExecutor.getConnectionManager().getExecutor().execute(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
while (true) {
|
||||
Runnable runnable = operations.poll();
|
||||
if (runnable != null) {
|
||||
runnable.run();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
sent.set(false);
|
||||
if (!operations.isEmpty()) {
|
||||
enqueueRunnable(null);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -1,63 +0,0 @@
|
||||
/**
|
||||
* 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.concurrent.ExecutorService;
|
||||
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.command.CommandAsyncExecutor;
|
||||
|
||||
import io.netty.util.concurrent.Future;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <R> result type
|
||||
*/
|
||||
public abstract class MapWriterExecutorPromise<R> extends MapWriterPromise<R> {
|
||||
|
||||
public MapWriterExecutorPromise(RFuture<R> f, CommandAsyncExecutor commandExecutor) {
|
||||
super(f, commandExecutor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(final Future<R> future, ExecutorService executorService) {
|
||||
if (condition(future)) {
|
||||
executorService.execute(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
executeWriter();
|
||||
} catch (Exception e) {
|
||||
tryFailure(e);
|
||||
return;
|
||||
}
|
||||
trySuccess(future.getNow());
|
||||
}
|
||||
});
|
||||
} else {
|
||||
trySuccess(future.getNow());
|
||||
}
|
||||
}
|
||||
|
||||
protected boolean condition(Future<R> future) {
|
||||
return true;
|
||||
}
|
||||
|
||||
protected abstract void executeWriter();
|
||||
|
||||
}
|
||||
@ -0,0 +1,34 @@
|
||||
/**
|
||||
* 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 io.netty.util.concurrent.Future;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <R> return type
|
||||
*/
|
||||
public abstract class MapWriterTask<R> {
|
||||
|
||||
protected boolean condition(Future<R> future) {
|
||||
return true;
|
||||
}
|
||||
|
||||
protected abstract void execute();
|
||||
|
||||
}
|
||||
@ -0,0 +1,110 @@
|
||||
/**
|
||||
* 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;
|
||||
|
||||
import org.redisson.api.map.MapLoader;
|
||||
import org.redisson.api.map.MapWriter;
|
||||
|
||||
/**
|
||||
* RMap options object.
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
public class MapOptions<K, V> {
|
||||
|
||||
public enum WriteMode {
|
||||
|
||||
/**
|
||||
* In write behind mode all data written in map object
|
||||
* also written using MapWriter in asynchronous mode.
|
||||
*/
|
||||
WRITE_BEHIND,
|
||||
|
||||
/**
|
||||
* In write through mode all write operations for map object
|
||||
* are synchronized with MapWriter write operations.
|
||||
* If MapWriter throws an error then it will be re-thrown to Map operation caller.
|
||||
*/
|
||||
WRITE_THROUGH
|
||||
|
||||
}
|
||||
|
||||
private MapLoader<K, V> loader;
|
||||
private MapWriter<K, V> writer;
|
||||
private WriteMode writeMode;
|
||||
|
||||
protected MapOptions() {
|
||||
}
|
||||
|
||||
protected MapOptions(MapOptions<K, V> copy) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new instance of MapOptions with default options.
|
||||
* <p>
|
||||
* This is equivalent to:
|
||||
* <pre>
|
||||
* new MapOptions()
|
||||
* .writer(null, null).loader(null);
|
||||
* </pre>
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*
|
||||
* @return MapOptions instance
|
||||
*
|
||||
*/
|
||||
public static <K, V> MapOptions<K, V> defaults() {
|
||||
return new MapOptions<K, V>();
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets map writer object used for write-through operations.
|
||||
*
|
||||
* @param writer object
|
||||
* @param writeMode for writer
|
||||
* @return MapOptions instance
|
||||
*/
|
||||
public MapOptions<K, V> writer(MapWriter<K, V> writer, WriteMode writeMode) {
|
||||
this.writer = writer;
|
||||
this.writeMode = writeMode;
|
||||
return this;
|
||||
}
|
||||
public MapWriter<K, V> getWriter() {
|
||||
return writer;
|
||||
}
|
||||
public WriteMode getWriteMode() {
|
||||
return writeMode;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets map reader object used for write-through operations.
|
||||
*
|
||||
* @param loader object
|
||||
* @return MapOptions instance
|
||||
*/
|
||||
public MapOptions<K, V> loader(MapLoader<K, V> loader) {
|
||||
this.loader = loader;
|
||||
return this;
|
||||
}
|
||||
public MapLoader<K, V> getLoader() {
|
||||
return loader;
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue