部分场景下 watcher 回调不成功

[lzbht]

1. 新增配置 -> 触发成功
2. 删除配置 -> 触发成功
3. 修改配置 -> 偶数次修改成功 例如连续修改三次 a = 1 / a = 12 / a = 123 其中 只有 a = 12 会触发回调

[lzbht]

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import asyncio
import threading

from nacos import NacosClient

from utils.environment_config import get_env_settings
from utils.logger import logger
from utils.public import singleton


@singleton
class ConfigurationService(object):
    def __init__(self):
        self.config = None
        self.env_settings = get_env_settings()
        print(self.env_settings)
        self.client = NacosClient(
            server_addresses=self.env_settings.nacos_address,
            namespace=self.env_settings.nacos_namespace,
        )
        self.thread = threading.Thread(target=self._send_heartbeat_in_thread)
        self.data = {}
        self.heart_beat_interval: int = 5

    def start(self):
        self.thread.start()
        self.add_watcher()
        self.fetch_and_apply_config()

    def get(self, key: str):
        return self.data.get(key)

    async def _send_heartbeat(self, service_name, ip, port):
        while True:
            try:
                await asyncio.to_thread(lambda: self.client.send_heartbeat(service_name, ip, port))
            except Exception as e:
                logger.error(f"Failed to send heartbeat: {str(e)}")
            await asyncio.sleep(self.heart_beat_interval)

    def _send_heartbeat_in_thread(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(
            self._send_heartbeat(
                self.env_settings.nacos_server_name,
                self.env_settings.nacos_server_ip,
                self.env_settings.nacos_server_port,
            )
        )

    def _read_application_properties(self, config):
        print("read something")
        config = config.encode().decode("unicode_escape")
        lines = config.splitlines()
        for line in lines:
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                key, value = line.split("=", 1)
                self.data[key.strip()] = value.strip()

    def fetch_and_apply_config(self):
        """
        Fetches configuration from the Nacos server and applies application properties.
        :return: None
        """
        config = self.client.get_config(
            self.env_settings.nacos_data_id, self.env_settings.nacos_group
        )
        self._read_application_properties(config)

    def add_watcher(self):
        self.client.add_config_watcher(
            self.env_settings.nacos_data_id, self.env_settings.nacos_group, self._read_application_properties
        )