последовательность
В этой статье в основном изучается DelayedQueue Redisson.
maven
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.8.1</version>
</dependency>
Пример
@Test
public void testDelayedQueue() throws InterruptedException {
Config config = new Config();
config.useSingleServer()
.setAddress("redis://192.168.99.100:6379");
RedissonClient redisson = Redisson.create(config);
RBlockingQueue<String> blockingQueue = redisson.getBlockingQueue("dest_queue1");
RDelayedQueue<String> delayedQueue = redisson.getDelayedQueue(blockingQueue);
delayedQueue.offer("demo", 10, TimeUnit.SECONDS);
Assert.assertFalse(blockingQueue.contains("demo"));
TimeUnit.SECONDS.sleep(15);
Assert.assertTrue(blockingQueue.contains("demo"));
}
- Здесь используются две очереди, операция offer delayedQueue напрямую вводится в delayedQueue, но delay действует на целевую очередь, здесь RBlockingQueue
Анализ исходного кода
RDelayedQueue.offer
redisson-3.8.1-sources.jar!/org/redisson/RedissonDelayedQueue.java
public class RedissonDelayedQueue<V> extends RedissonExpirable implements RDelayedQueue<V> {
private final QueueTransferService queueTransferService;
private final String channelName;
private final String queueName;
private final String timeoutSetName;
protected RedissonDelayedQueue(QueueTransferService queueTransferService, Codec codec, final CommandAsyncExecutor commandExecutor, String name) {
super(codec, commandExecutor, name);
channelName = prefixName("redisson_delay_queue_channel", getName());
queueName = prefixName("redisson_delay_queue", getName());
timeoutSetName = prefixName("redisson_delay_queue_timeout", getName());
//QueueTransferTask task = ......
queueTransferService.schedule(queueName, task);
this.queueTransferService = queueTransferService;
}
public void offer(V e, long delay, TimeUnit timeUnit) {
get(offerAsync(e, delay, timeUnit));
}
public RFuture<Void> offerAsync(V e, long delay, TimeUnit timeUnit) {
long delayInMs = timeUnit.toMillis(delay);
long timeout = System.currentTimeMillis() + delayInMs;
long randomId = PlatformDependent.threadLocalRandom().nextLong();
return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_VOID,
"local value = struct.pack('dLc0', tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);"
+ "redis.call('zadd', KEYS[2], ARGV[1], value);"
+ "redis.call('rpush', KEYS[3], value);"
// if new object added to queue head when publish its startTime
// to all scheduler workers
+ "local v = redis.call('zrange', KEYS[2], 0, 0); "
+ "if v[1] == value then "
+ "redis.call('publish', KEYS[4], ARGV[1]); "
+ "end;"
,
Arrays.<Object>asList(getName(), timeoutSetName, queueName, channelName),
timeout, randomId, encode(e));
}
public ByteBuf encode(Object value) {
if (commandExecutor.isRedissonReferenceSupportEnabled()) {
RedissonReference reference = RedissonObjectFactory.toReference(commandExecutor.getConnectionManager().getCfg(), value);
if (reference != null) {
value = reference;
}
}
try {
return codec.getValueEncoder().encode(value);
} catch (IOException e) {
throw new IllegalArgumentException(e);
}
}
public static String prefixName(String prefix, String name) {
if (name.contains("{")) {
return prefix + ":" + name;
}
return prefix + ":{" + name + "}";
}
//......
}
- Вот сценарий lua, где массив параметров ключей имеет четыре значения, KEYS[1] — это getName(), KEYS[2] — это timeoutSetName, KEYS[3] — это queueName, KEYS[4] — это channelName
- Есть три переменные: ARGV[1] — тайм-аут, ARGV[2] — randomId, ARGV[3] — encode(e)
- Этот lua-скрипт добавляет структуру в zset timeoutSetName, счет которой является значением тайм-аута; добавляет структуру в конец списка queueName; затем оценивает, является ли первый элемент zset timeoutSetName текущей структурой, если да, затем опубликуйте сообщение о тайм-ауте на канале
queueTransferService.schedule
redisson-3.8.1-sources.jar!/org/redisson/RedissonDelayedQueue.java
QueueTransferTask task = new QueueTransferTask(commandExecutor.getConnectionManager()) {
@Override
protected RFuture<Long> pushTaskAsync() {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
"local expiredValues = redis.call('zrangebyscore', KEYS[2], 0, ARGV[1], 'limit', 0, ARGV[2]); "
+ "if #expiredValues > 0 then "
+ "for i, v in ipairs(expiredValues) do "
+ "local randomId, value = struct.unpack('dLc0', v);"
+ "redis.call('rpush', KEYS[1], value);"
+ "redis.call('lrem', KEYS[3], 1, v);"
+ "end; "
+ "redis.call('zrem', KEYS[2], unpack(expiredValues));"
+ "end; "
// get startTime from scheduler queue head task
+ "local v = redis.call('zrange', KEYS[2], 0, 0, 'WITHSCORES'); "
+ "if v[1] ~= nil then "
+ "return v[2]; "
+ "end "
+ "return nil;",
Arrays.<Object>asList(getName(), timeoutSetName, queueName),
System.currentTimeMillis(), 100);
}
@Override
protected RTopic<Long> getTopic() {
return new RedissonTopic<Long>(LongCodec.INSTANCE, commandExecutor, channelName);
}
};
queueTransferService.schedule(queueName, task);
- QueueTransferTask запланирован в конструкторе RedissonDelayedQueue.
- Выполнение планирования — это метод pushTaskAsync, который в основном перемещает элементы с истекшим сроком действия из очереди элементов в целевую очередь.
- Здесь используется сценарий lua, KEYS[1] — это getName(), KEYS[2] — это timeoutSetName, KEYS[3] — это queueName, ARGV[1] — текущая отметка времени, а ARGV[2] — 100.
- Здесь вызывается zrangebyscore, и zset timeoutSetName сортируется с использованием параметра timeout, и получаются элементы, точки которых находятся между 0 и текущей отметкой времени, и берутся первые 200 элементов.
- Если есть значение, указывающее, что элемент необходимо передать в целевую очередь, то вызовите rpush, чтобы передать его в целевую очередь, затем вызовите lrem, чтобы удалить его из очереди элементов, и, наконец, удалите обработанные элементы из zset. timeoutSetName
- После обработки передачи элемента возьмите оценку первого элемента zset из timeoutSetName и верните ее, если не верните nil
QueueTransferService.schedule
redisson-3.8.1-sources.jar!/org/redisson/QueueTransferService.java
public class QueueTransferService {
private final ConcurrentMap<String, QueueTransferTask> tasks = PlatformDependent.newConcurrentHashMap();
public synchronized void schedule(String name, QueueTransferTask task) {
QueueTransferTask oldTask = tasks.putIfAbsent(name, task);
if (oldTask == null) {
task.start();
} else {
oldTask.incUsage();
}
}
public synchronized void remove(String name) {
QueueTransferTask task = tasks.get(name);
if (task != null) {
if (task.decUsage() == 0) {
tasks.remove(name, task);
task.stop();
}
}
}
}
- Метод расписания здесь сначала добавляется в ConcurrentMap, если задача уже существует, вызываем oldTask.incUsage(), если не существует, запускаем задачу
QueueTransferTask.start
redisson-3.8.1-sources.jar!/org/redisson/QueueTransferTask.java
public void start() {
RTopic<Long> schedulerTopic = getTopic();
statusListenerId = schedulerTopic.addListener(new BaseStatusListener() {
@Override
public void onSubscribe(String channel) {
pushTask();
}
});
messageListenerId = schedulerTopic.addListener(new MessageListener<Long>() {
@Override
public void onMessage(CharSequence channel, Long startTime) {
scheduleTask(startTime);
}
});
}
private void scheduleTask(final Long startTime) {
TimeoutTask oldTimeout = lastTimeout.get();
if (startTime == null) {
return;
}
if (oldTimeout != null) {
oldTimeout.getTask().cancel();
}
long delay = startTime - System.currentTimeMillis();
if (delay > 10) {
Timeout timeout = connectionManager.newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
pushTask();
TimeoutTask currentTimeout = lastTimeout.get();
if (currentTimeout.getTask() == timeout) {
lastTimeout.compareAndSet(currentTimeout, null);
}
}
}, delay, TimeUnit.MILLISECONDS);
if (!lastTimeout.compareAndSet(oldTimeout, new TimeoutTask(startTime, timeout))) {
timeout.cancel();
}
} else {
pushTask();
}
}
private void pushTask() {
RFuture<Long> startTimeFuture = pushTaskAsync();
startTimeFuture.addListener(new FutureListener<Long>() {
@Override
public void operationComplete(io.netty.util.concurrent.Future<Long> future) throws Exception {
if (!future.isSuccess()) {
if (future.cause() instanceof RedissonShutdownException) {
return;
}
log.error(future.cause().getMessage(), future.cause());
scheduleTask(System.currentTimeMillis() + 5 * 1000L);
return;
}
if (future.getNow() != null) {
scheduleTask(future.getNow());
}
}
});
}
- Здесь используется RTopic, добавлены StatusListener и MessageListener
- StatusListener запускает pushTask при подписке, а MessageListener в основном вызывает scheduleTask.
- Реализация pushTaskAsync в RedissonDelayedQueue — это передача упомянутых выше элементов реализации в исходную очередь и целевую очередь
- Метод scheduleTask будет пересчитывать задержку: при задержке больше 10 срабатывает pushTask, а если задержка меньше или равна 10, pushTask срабатывает немедленно.
- pushTask вызовет операцию pushTaskAsync. Если выполнение не удалось, scheduleTask будет запущен повторно. Если выполнение выполнено успешно, но возвращаемое значение (
timeoutSetName的zset的第一个元素的得分) не равно нулю, вызовите scheduleTask с этим значением
резюме
- DelayedQueue redisson используется для постановки в очередь элементов и задержки информации, а затем запланированная задача передает элементы с истекшим сроком действия в целевую очередь.
- Здесь для хранения используются три структуры: одна — целевой список очередей; одна — собственный список очередей, который добавляет структуру с информацией о задержке; одна — zset timeoutSetName, элемент — это структура, а ее оценка — это значение тайм-аута. .
- Redisson использует много асинхронных обратных вызовов для работы, и общее чтение кода будет относительно трудоемким.