diff --git a/api/schedule/queue_monitor_task.py b/api/schedule/queue_monitor_task.py
index a05e1358e..4d517e549 100644
--- a/api/schedule/queue_monitor_task.py
+++ b/api/schedule/queue_monitor_task.py
@@ -1,8 +1,8 @@
 import logging
 from datetime import datetime
-from urllib.parse import urlparse
 
 import click
+from kombu.utils.url import parse_url  # type: ignore
 from redis import Redis
 
 import app
@@ -10,16 +10,13 @@ from configs import dify_config
 from extensions.ext_database import db
 from libs.email_i18n import EmailType, get_email_i18n_service
 
-# Create a dedicated Redis connection (using the same configuration as Celery)
-celery_broker_url = dify_config.CELERY_BROKER_URL
-
-parsed = urlparse(celery_broker_url)
-host = parsed.hostname or "localhost"
-port = parsed.port or 6379
-password = parsed.password or None
-redis_db = parsed.path.strip("/") or "1"  # type: ignore
-
-celery_redis = Redis(host=host, port=port, password=password, db=redis_db)
+redis_config = parse_url(dify_config.CELERY_BROKER_URL)
+celery_redis = Redis(
+    host=redis_config["hostname"],
+    port=redis_config["port"],
+    password=redis_config["password"],
+    db=int(redis_config["virtual_host"]) if redis_config["virtual_host"] else 1,
+)
 
 
 @app.celery.task(queue="monitor")
diff --git a/api/tests/unit_tests/configs/test_dify_config.py b/api/tests/unit_tests/configs/test_dify_config.py
index e9d4ee193..0ae6a09f5 100644
--- a/api/tests/unit_tests/configs/test_dify_config.py
+++ b/api/tests/unit_tests/configs/test_dify_config.py
@@ -1,5 +1,6 @@
 import os
 
+import pytest
 from flask import Flask
 from packaging.version import Version
 from yarl import URL
@@ -137,3 +138,61 @@ def test_db_extras_options_merging(monkeypatch):
     options = engine_options["connect_args"]["options"]
     assert "search_path=myschema" in options
     assert "timezone=UTC" in options
+
+
+@pytest.mark.parametrize(
+    ("broker_url", "expected_host", "expected_port", "expected_username", "expected_password", "expected_db"),
+    [
+        ("redis://localhost:6379/1", "localhost", 6379, None, None, "1"),
+        ("redis://:password@localhost:6379/1", "localhost", 6379, None, "password", "1"),
+        ("redis://:mypass%23123@localhost:6379/1", "localhost", 6379, None, "mypass#123", "1"),
+        ("redis://user:pass%40word@redis-host:6380/2", "redis-host", 6380, "user", "pass@word", "2"),
+        ("redis://admin:complex%23pass%40word@127.0.0.1:6379/0", "127.0.0.1", 6379, "admin", "complex#pass@word", "0"),
+        (
+            "redis://user%40domain:secret%23123@redis.example.com:6380/3",
+            "redis.example.com",
+            6380,
+            "user@domain",
+            "secret#123",
+            "3",
+        ),
+        # Password containing %23 substring (double encoding scenario)
+        ("redis://:mypass%2523@localhost:6379/1", "localhost", 6379, None, "mypass%23", "1"),
+        # Username and password both containing encoded characters
+        ("redis://user%2525%40:pass%2523@localhost:6379/1", "localhost", 6379, "user%25@", "pass%23", "1"),
+    ],
+)
+def test_celery_broker_url_with_special_chars_password(
+    monkeypatch, broker_url, expected_host, expected_port, expected_username, expected_password, expected_db
+):
+    """Test that CELERY_BROKER_URL with various formats are handled correctly."""
+    from kombu.utils.url import parse_url
+
+    # clear system environment variables
+    os.environ.clear()
+
+    # Set up basic required environment variables (following existing pattern)
+    monkeypatch.setenv("CONSOLE_API_URL", "https://example.com")
+    monkeypatch.setenv("CONSOLE_WEB_URL", "https://example.com")
+    monkeypatch.setenv("DB_USERNAME", "postgres")
+    monkeypatch.setenv("DB_PASSWORD", "postgres")
+    monkeypatch.setenv("DB_HOST", "localhost")
+    monkeypatch.setenv("DB_PORT", "5432")
+    monkeypatch.setenv("DB_DATABASE", "dify")
+
+    # Set the CELERY_BROKER_URL to test
+    monkeypatch.setenv("CELERY_BROKER_URL", broker_url)
+
+    # Create config and verify the URL is stored correctly
+    config = DifyConfig()
+    assert broker_url == config.CELERY_BROKER_URL
+
+    # Test actual parsing behavior using kombu's parse_url (same as production)
+    redis_config = parse_url(config.CELERY_BROKER_URL)
+
+    # Verify the parsing results match expectations (using kombu's field names)
+    assert redis_config["hostname"] == expected_host
+    assert redis_config["port"] == expected_port
+    assert redis_config["userid"] == expected_username  # kombu uses 'userid' not 'username'
+    assert redis_config["password"] == expected_password
+    assert redis_config["virtual_host"] == expected_db  # kombu uses 'virtual_host' not 'db'