redis execute `rpop` command got block about redis-rs HOT 10 CLOSED

use redis::AsyncCommands;
use tracing::level_filters::LevelFilter;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

struct Sender {
    redis_client: redis::Client,
    queue: String,
}

impl Sender {
    fn init(redis_addr: &str) -> anyhow::Result<Self> {
        let redis_client = redis::Client::open(redis_addr)?;
        Ok(Sender {
            redis_client,
            queue: "test-queue".to_owned(),
        })
    }

    fn get_redis_client(&self) -> redis::Client {
        self.redis_client.clone()
    }

    async fn save_to_redis(&self, data: String) -> anyhow::Result<()> {
        let client = self.get_redis_client();
        tracing::info!("Get redis connection");
        let mut conn = client.get_multiplexed_async_connection().await?;
        tracing::info!("Execute save command");
        conn.lpush(&self.queue, data).await?;
        tracing::info!("Save to redis succeed");

        Ok(())
    }

    async fn run(self) -> anyhow::Result<()> {
        self.save_to_redis("Test Data".to_owned()).await?;
        let client = self.get_redis_client();
        let mut conn = client.get_multiplexed_async_connection().await?;
        loop {
            tracing::info!("Fetching redis data");
            let data: String = conn.rpop(&self.queue, None).await?;
            if data.is_empty() {
                tracing::error!("Data exception");
                break;
            }
            tracing::info!("Data fetch succeed");
            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
            self.save_to_redis(data).await?;
            tracing::info!("Finished once data submit");
        }
        Ok(())
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::registry()
        .with(LevelFilter::INFO)
        .with(tracing_subscriber::fmt::layer())
        .init();
    let redis_url = "redis://remote-server/";
    let sender = Sender::init(redis_url)?;
    tokio::spawn(sender.run());
    tracing::info!("Thread started");
    loop {
        tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::registry()
        .with(LevelFilter::INFO)
        .with(tracing_subscriber::fmt::layer())
        .init();
    let redis_url = "redis://remote-server";
    let sender = Sender::init(redis_url)?;
    sender.run().await?;
    Ok(())
}