# 將一個(gè)或多個(gè)值 value 插入到列表 key 的表頭
LPUSH key value [value …]

# 阻塞式等待，將列表 source 中的最后一個(gè)元素 (尾元素) 彈出，并返回給客戶端。將 source 彈出的元素插入到列表 destination ，作為 destination 列表的的頭元素。超時(shí)參數(shù) timeout 接受一個(gè)以秒為單位的數(shù)字作為值。超時(shí)參數(shù)設(shè)為 0 表示阻塞時(shí)間可以無(wú)限期延長(zhǎng) (block indefinitely) 。
BRPOPLPUSH source destination timeout

# 根據(jù)參數(shù) count 的值，移除列表中與參數(shù) value 相等的元素。
LREM key count value

@Resource
private RedisConnectionFactory connectionFactory;

public void lPush(@Nonnull String key, @Nonnull String value) {
  RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory);
  try {
    byte[] byteKey = RedisSerializer.string().serialize(getKey(key));
    byte[] byteValue = RedisSerializer.string().serialize(value);
    assert byteKey != null;
    connection.lPush(byteKey, byteValue);
  } finally {
    RedisConnectionUtils.releaseConnection(connection, connectionFactory);
  }
}

public void bRPopLPush(@Nonnull String key, Consumer<String> consumer) {
  CompletableFuture.runAsync(() -> {
    RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory);
    try {
      byte[] srcKey = RedisSerializer.string().serialize(getKey(key));
      byte[] dstKey = RedisSerializer.string().serialize(getBackupKey(key));
      assert srcKey != null;
      assert dstKey != null;
      while (true) {
        byte[] byteValue = new byte[0];
        boolean success = false;
        try {
          byteValue = connection.bRPopLPush(0, srcKey, dstKey);
          if (byteValue != null && byteValue.length != 0) {
            consumer.accept(new String(byteValue));
            success = true;
          }
        } catch (Exception ignored) {
          // 防止獲取 key 達(dá)到超時(shí)時(shí)間拋出 QueryTimeoutException 異常退出
        } finally {
          if (success) {
            // 處理成功才刪除備份隊(duì)列的 key
            connection.lRem(dstKey, 1, byteValue);
          }
        }
      }
    } finally {
      RedisConnectionUtils.releaseConnection(connection, connectionFactory);
    }
  });
}

@Test
public void testLPush() throws InterruptedException {
  String queueA = "queueA";
  int i = 0;
  while (true) {
    String msg = "Hello-" + i++;
    redisBlockQueue.lPush(queueA, msg);
    System.out.println("lPush: " + msg);
    Thread.sleep(3000);
  }
}

@Test
public void testBRPopLPush() {
  String queueA = "queueA";
  redisBlockQueue.bRPopLPush(queueA, (val) -> {
    // 在這里處理具體的業(yè)務(wù)邏輯
    System.out.println("val: " + val);
  });

  // 防止 Junit 進(jìn)程退出
  LockSupport.park();
}

@Resource
private RedisBlockQueue redisBlockQueue;

@PostConstruct
public void init() {
   redisBlockQueue.bRPopLPush(xx, (value) -> {
     //...
   });
}