diff --git a/Makefile b/Makefile
index 2c9798b8a7..5165ad577b 100644
--- a/Makefile
+++ b/Makefile
@@ -238,6 +238,12 @@ receiver:
 	fi; \
 	$(PYTHON) manage.py run_callback_receiver
 
+rsyslog-configurer:
+	@if [ "$(VENV_BASE)" ]; then \
+		. $(VENV_BASE)/awx/bin/activate; \
+	fi; \
+	$(PYTHON) manage.py run_rsyslog_configurer
+
 nginx:
 	nginx -g "daemon off;"
 
diff --git a/awx/conf/tests/functional/test_api.py b/awx/conf/tests/functional/test_api.py
index 2d09423386..b600c3766d 100644
--- a/awx/conf/tests/functional/test_api.py
+++ b/awx/conf/tests/functional/test_api.py
@@ -94,9 +94,7 @@ def test_setting_singleton_retrieve_readonly(api_request, dummy_setting):
 
 @pytest.mark.django_db
 def test_setting_singleton_update(api_request, dummy_setting):
-    with dummy_setting('FOO_BAR', field_class=fields.IntegerField, category='FooBar', category_slug='foobar'), mock.patch(
-        'awx.conf.views.handle_setting_changes'
-    ):
+    with dummy_setting('FOO_BAR', field_class=fields.IntegerField, category='FooBar', category_slug='foobar'), mock.patch('awx.conf.views.clear_setting_cache'):
         api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), data={'FOO_BAR': 3})
         response = api_request('get', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
         assert response.data['FOO_BAR'] == 3
@@ -112,7 +110,7 @@ def test_setting_singleton_update_hybriddictfield_with_forbidden(api_request, du
     # sure that the _Forbidden validator doesn't get used for the
     # fields.  See also https://github.com/ansible/awx/issues/4099.
     with dummy_setting('FOO_BAR', field_class=sso_fields.SAMLOrgAttrField, category='FooBar', category_slug='foobar'), mock.patch(
-        'awx.conf.views.handle_setting_changes'
+        'awx.conf.views.clear_setting_cache'
     ):
         api_request(
             'patch',
@@ -126,7 +124,7 @@ def test_setting_singleton_update_hybriddictfield_with_forbidden(api_request, du
 @pytest.mark.django_db
 def test_setting_singleton_update_dont_change_readonly_fields(api_request, dummy_setting):
     with dummy_setting('FOO_BAR', field_class=fields.IntegerField, read_only=True, default=4, category='FooBar', category_slug='foobar'), mock.patch(
-        'awx.conf.views.handle_setting_changes'
+        'awx.conf.views.clear_setting_cache'
     ):
         api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), data={'FOO_BAR': 5})
         response = api_request('get', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
@@ -136,7 +134,7 @@ def test_setting_singleton_update_dont_change_readonly_fields(api_request, dummy
 @pytest.mark.django_db
 def test_setting_singleton_update_dont_change_encrypted_mark(api_request, dummy_setting):
     with dummy_setting('FOO_BAR', field_class=fields.CharField, encrypted=True, category='FooBar', category_slug='foobar'), mock.patch(
-        'awx.conf.views.handle_setting_changes'
+        'awx.conf.views.clear_setting_cache'
     ):
         api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), data={'FOO_BAR': 'password'})
         assert Setting.objects.get(key='FOO_BAR').value.startswith('$encrypted$')
@@ -155,16 +153,14 @@ def test_setting_singleton_update_runs_custom_validate(api_request, dummy_settin
 
     with dummy_setting('FOO_BAR', field_class=fields.IntegerField, category='FooBar', category_slug='foobar'), dummy_validate(
         'foobar', func_raising_exception
-    ), mock.patch('awx.conf.views.handle_setting_changes'):
+    ), mock.patch('awx.conf.views.clear_setting_cache'):
         response = api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), data={'FOO_BAR': 23})
         assert response.status_code == 400
 
 
 @pytest.mark.django_db
 def test_setting_singleton_delete(api_request, dummy_setting):
-    with dummy_setting('FOO_BAR', field_class=fields.IntegerField, category='FooBar', category_slug='foobar'), mock.patch(
-        'awx.conf.views.handle_setting_changes'
-    ):
+    with dummy_setting('FOO_BAR', field_class=fields.IntegerField, category='FooBar', category_slug='foobar'), mock.patch('awx.conf.views.clear_setting_cache'):
         api_request('delete', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
         response = api_request('get', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
         assert not response.data['FOO_BAR']
@@ -173,7 +169,7 @@ def test_setting_singleton_delete(api_request, dummy_setting):
 @pytest.mark.django_db
 def test_setting_singleton_delete_no_read_only_fields(api_request, dummy_setting):
     with dummy_setting('FOO_BAR', field_class=fields.IntegerField, read_only=True, default=23, category='FooBar', category_slug='foobar'), mock.patch(
-        'awx.conf.views.handle_setting_changes'
+        'awx.conf.views.clear_setting_cache'
     ):
         api_request('delete', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
         response = api_request('get', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
diff --git a/awx/conf/views.py b/awx/conf/views.py
index 5231cd86bd..e4a8375122 100644
--- a/awx/conf/views.py
+++ b/awx/conf/views.py
@@ -26,10 +26,11 @@ from awx.api.generics import APIView, GenericAPIView, ListAPIView, RetrieveUpdat
 from awx.api.permissions import IsSystemAdminOrAuditor
 from awx.api.versioning import reverse
 from awx.main.utils import camelcase_to_underscore
-from awx.main.tasks.system import handle_setting_changes
+from awx.main.tasks.system import clear_setting_cache
 from awx.conf.models import Setting
 from awx.conf.serializers import SettingCategorySerializer, SettingSingletonSerializer
 from awx.conf import settings_registry
+from awx.main.utils.external_logging import send_pg_notify
 
 
 SettingCategory = collections.namedtuple('SettingCategory', ('url', 'slug', 'name'))
@@ -118,7 +119,10 @@ class SettingSingletonDetail(RetrieveUpdateDestroyAPIView):
                 setting.save(update_fields=['value'])
                 settings_change_list.append(key)
         if settings_change_list:
-            connection.on_commit(lambda: handle_setting_changes.delay(settings_change_list))
+            connection.on_commit(lambda: clear_setting_cache.delay(settings_change_list))
+            if any([setting.startswith('LOG_AGGREGATOR') for setting in settings_change_list]):
+                # call notify to rsyslog. no data is need so payload is empty
+                send_pg_notify('rsyslog_configurer', "")
 
     def destroy(self, request, *args, **kwargs):
         instance = self.get_object()
@@ -133,7 +137,10 @@ class SettingSingletonDetail(RetrieveUpdateDestroyAPIView):
             setting.delete()
             settings_change_list.append(setting.key)
         if settings_change_list:
-            connection.on_commit(lambda: handle_setting_changes.delay(settings_change_list))
+            connection.on_commit(lambda: clear_setting_cache.delay(settings_change_list))
+            if any([setting.startswith('LOG_AGGREGATOR') for setting in settings_change_list]):
+                # call notify to rsyslog. no data is need so payload is empty
+                send_pg_notify('rsyslog_configurer', "")
 
         # When TOWER_URL_BASE is deleted from the API, reset it to the hostname
         # used to make the request as a default.
diff --git a/awx/main/management/commands/run_rsyslog_configurer.py b/awx/main/management/commands/run_rsyslog_configurer.py
new file mode 100644
index 0000000000..5cc214ced9
--- /dev/null
+++ b/awx/main/management/commands/run_rsyslog_configurer.py
@@ -0,0 +1,38 @@
+import logging
+
+from django.core.management.base import BaseCommand
+from django.conf import settings
+from django.core.cache import cache
+from awx.main.dispatch import pg_bus_conn
+from awx.main.utils.external_logging import reconfigure_rsyslog
+
+logger = logging.getLogger('awx.main.rsyslog_configurer')
+
+
+class Command(BaseCommand):
+    """
+    Rsyslog Configurer
+    Runs as a management command and starts rsyslog configurer daemon. Daemon listens
+    for pg_notify then calls reconfigure_rsyslog
+    """
+
+    help = 'Launch the rsyslog_configurer daemon'
+
+    def handle(self, *arg, **options):
+        try:
+            with pg_bus_conn(new_connection=True) as conn:
+                conn.listen("rsyslog_configurer")
+                # reconfigure rsyslog on start up
+                reconfigure_rsyslog()
+                for e in conn.events(yield_timeouts=True):
+                    if e is not None:
+                        logger.info("Change in logging settings found. Restarting rsyslogd")
+                        # clear the cache of relevant settings then restart
+                        setting_keys = [k for k in dir(settings) if k.startswith('LOG_AGGREGATOR')]
+                        cache.delete_many(setting_keys)
+                        settings._awx_conf_memoizedcache.clear()
+                        reconfigure_rsyslog()
+        except Exception:
+            # Log unanticipated exception in addition to writing to stderr to get timestamps and other metadata
+            logger.exception('Encountered unhandled error in rsyslog_configurer main loop')
+            raise
diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py
index ddff6b4ec4..ac0e0c32bf 100644
--- a/awx/main/tasks/system.py
+++ b/awx/main/tasks/system.py
@@ -59,7 +59,6 @@ from awx.main.utils.common import (
     ScheduleTaskManager,
 )
 
-from awx.main.utils.external_logging import reconfigure_rsyslog
 from awx.main.utils.reload import stop_local_services
 from awx.main.utils.pglock import advisory_lock
 from awx.main.tasks.receptor import get_receptor_ctl, worker_info, worker_cleanup, administrative_workunit_reaper, write_receptor_config
@@ -115,9 +114,6 @@ def dispatch_startup():
     m = Metrics()
     m.reset_values()
 
-    # Update Tower's rsyslog.conf file based on loggins settings in the db
-    reconfigure_rsyslog()
-
 
 def inform_cluster_of_shutdown():
     try:
@@ -245,7 +241,7 @@ def apply_cluster_membership_policies():
 
 
 @task(queue='tower_broadcast_all')
-def handle_setting_changes(setting_keys):
+def clear_setting_cache(setting_keys):
     orig_len = len(setting_keys)
     for i in range(orig_len):
         for dependent_key in settings_registry.get_dependent_settings(setting_keys[i]):
@@ -254,9 +250,6 @@ def handle_setting_changes(setting_keys):
     logger.debug('cache delete_many(%r)', cache_keys)
     cache.delete_many(cache_keys)
 
-    if any([setting.startswith('LOG_AGGREGATOR') for setting in setting_keys]):
-        reconfigure_rsyslog()
-
 
 @task(queue='tower_broadcast_all')
 def delete_project_files(project_path):
diff --git a/awx/main/tests/functional/api/test_settings.py b/awx/main/tests/functional/api/test_settings.py
index a1ae7398a5..bcfa499af3 100644
--- a/awx/main/tests/functional/api/test_settings.py
+++ b/awx/main/tests/functional/api/test_settings.py
@@ -279,7 +279,7 @@ def test_logging_aggregator_missing_settings(put, post, admin, key, value, error
     ],
 )
 @pytest.mark.django_db
-def test_logging_aggregator_valid_settings(put, post, admin, type, host, port, username, password):
+def test_logging_aggregator_valid_settings(put, post, admin, type, host, port, username, password, mocker):
     _, mock_settings = _mock_logging_defaults()
     # type = 'splunk'
     # host = 'https://yoursplunk:8088/services/collector/event'
@@ -292,6 +292,8 @@ def test_logging_aggregator_valid_settings(put, post, admin, type, host, port, u
         mock_settings['LOG_AGGREGATOR_USERNAME'] = username
     if password:
         mock_settings['LOG_AGGREGATOR_PASSWORD'] = password
+    # mock testing pg_notify
+    mocker.patch("awx.conf.views.send_pg_notify", return_value=None)
     url = reverse('api:setting_singleton_detail', kwargs={'category_slug': 'logging'})
     response = put(url, data=mock_settings, user=admin, expect=200)
     assert type in response.data.get('LOG_AGGREGATOR_TYPE')
@@ -305,13 +307,15 @@ def test_logging_aggregator_valid_settings(put, post, admin, type, host, port, u
 
 
 @pytest.mark.django_db
-def test_logging_aggregator_connection_test_valid(put, post, admin):
+def test_logging_aggregator_connection_test_valid(put, post, admin, mocker):
     _, mock_settings = _mock_logging_defaults()
     type = 'other'
     host = 'https://localhost'
     mock_settings['LOG_AGGREGATOR_ENABLED'] = True
     mock_settings['LOG_AGGREGATOR_TYPE'] = type
     mock_settings['LOG_AGGREGATOR_HOST'] = host
+    # mock testing pg_notify
+    mocker.patch("awx.conf.views.send_pg_notify", return_value=None)
     # POST to save these mock settings
     url = reverse('api:setting_singleton_detail', kwargs={'category_slug': 'logging'})
     put(url, data=mock_settings, user=admin, expect=200)
diff --git a/awx/main/utils/external_logging.py b/awx/main/utils/external_logging.py
index 26f434a4e4..e3e0a3d1c8 100644
--- a/awx/main/utils/external_logging.py
+++ b/awx/main/utils/external_logging.py
@@ -6,6 +6,7 @@ import urllib.parse as urlparse
 from django.conf import settings
 
 from awx.main.utils.reload import supervisor_service_command
+from awx.main.dispatch import pg_bus_conn
 
 
 def construct_rsyslog_conf_template(settings=settings):
@@ -124,3 +125,8 @@ def reconfigure_rsyslog():
             f.write(tmpl + '\n')
         shutil.move(path, '/var/lib/awx/rsyslog/rsyslog.conf')
     supervisor_service_command(command='restart', service='awx-rsyslogd')
+
+
+def send_pg_notify(channel: str, payload: str) -> None:
+    with pg_bus_conn() as conn:
+        conn.notify(channel, payload)
diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py
index e7720e150b..6c1c765a27 100644
--- a/awx/settings/defaults.py
+++ b/awx/settings/defaults.py
@@ -867,6 +867,7 @@ LOGGING = {
         'awx.main.dispatch': {'handlers': ['dispatcher']},
         'awx.main.consumers': {'handlers': ['console', 'file', 'tower_warnings'], 'level': 'INFO'},
         'awx.main.wsbroadcast': {'handlers': ['wsbroadcast']},
+        'awx.main.rsyslog_configurer': {'handlers': ['rsyslog_configurer']},
         'awx.main.commands.inventory_import': {'handlers': ['inventory_import'], 'propagate': False},
         'awx.main.tasks': {'handlers': ['task_system', 'external_logger'], 'propagate': False},
         'awx.main.analytics': {'handlers': ['task_system', 'external_logger'], 'level': 'INFO', 'propagate': False},
@@ -897,6 +898,7 @@ handler_config = {
     'task_system': {'filename': 'task_system.log'},
     'rbac_migrations': {'filename': 'tower_rbac_migrations.log'},
     'job_lifecycle': {'filename': 'job_lifecycle.log', 'formatter': 'job_lifecycle'},
+    'rsyslog_configurer': {'filename': 'rsyslog_configurer.log'},
 }
 
 # If running on a VM, we log to files. When running in a container, we log to stdout.
diff --git a/tools/docker-compose/supervisor.conf b/tools/docker-compose/supervisor.conf
index 0e2441a47e..5107b463de 100644
--- a/tools/docker-compose/supervisor.conf
+++ b/tools/docker-compose/supervisor.conf
@@ -34,6 +34,16 @@ stdout_logfile_maxbytes=0
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 
+[program:awx-rsyslog-configurer]
+command = make rsyslog-configurer
+autorestart = true
+stopasgroup=true
+killasgroup=true
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+
 [program:awx-uwsgi]
 command = make uwsgi
 autorestart = true