Zookeeper C++编程实战之配置更新

缺乏、安全感 2022-04-17 03:36 302阅读 0赞

CZookeeperHelper：  
[https://github.com/eyjian/libmooon/blob/master/include/mooon/net/zookeeper\_helper.h][https_github.com_eyjian_libmooon_blob_master_include_mooon_net_zookeeper_helper.h]

CMainHelper:  
[https://github.com/eyjian/libmooon/blob/master/include/mooon/sys/main\_template.h][https_github.com_eyjian_libmooon_blob_master_include_mooon_sys_main_template.h]

// 演示一个多线程程序如何借助zookeeper，实现配置的动态更新
    //
    // 实现理念（有些场景不适合）：
    // 1) 让线程不涉及配置的动态更新，这样避免了动态更新配置
    // 2) 通过创建新线程的方式达到配置动态更新的目的，老的线程直接退出
    // 3) 先创建新线程，再退出老线程，保持服务不中断
    //
    // 实际上，也可以通过父子进程方式来达到配置动态更新，
    // 父进程检测到配置更新后，父进程读取配置，并检查配置的合法性。
    // 如果合法则创建新的子进程，完成后再kill原有的子进程，
    // 这样子进程就不涉及配置更新逻辑。
    //
    // 这两种方法，均可比较简单应对复杂的配置动态更新，
    // 但如果新旧配置无法同时兼容，则需要先停掉老的线程或进程，
    // 然后再启动新的线程或进程，否则做到无缝地动态更新。
    //
    // 编译要求环境：C++11或更高
    // 编译语句大致如下：
    // g++ -g -o b zk_conf_example.cpp -I/usr/local/mooon/include -I/usr/local/zookeeper/include /usr/local/mooon/lib/libmooon.a /usr/local/zookeeper/lib/libzookeeper_mt.a -pthread -std=c++11 -DMOOON_HAVE_ZOOKEEPER=1 -lz
    #include <mooon/net/zookeeper_helper.h>
    #include <mooon/sys/datetime_utils.h> // 格式化时间也可以考虑C++标准库提供的std::put_time
    #include <mooon/sys/main_template.h>
    #include <mooon/utils/args_parser.h>
    #include <chrono>
    #include <condition_variable>
    #include <mutex>
    #include <system_error>
    #include <thread>
    
    // 指定存放配置的zookeeper
    STRING_ARG_DEFINE(zookeeper, "", "Comma separated list of servers in the ZooKeeper Quorum, example: --zookeeper=127.0.0.1:2181");
    
    class CMyApplication;
    
    // 负责具体业务的工作者（线程）
    class CWorker
    {
    public:
        CWorker(CMyApplication* app, int index);
        void run(); // 线程入口函数
        void stop() { _stop = true; }
    
    private:
        CMyApplication* _app;
        int _index;
        volatile bool _stop;
    };
    
    // 应用程序主类（或叫上下文类，也可叫入口类）
    // 通过继承CZookeeperHelper，获得zookeeper操作能力，
    // 包括读写zookeeper数据能力、发现配置更新能力和主备切换能力。
    //
    // 可继承mooon::sys::CMainHelper，
    // 以获得通过信号SIGTERM的优雅退出能力，
    // CMainHelper提供了优雅和安全的信号处理，
    // 默认的优雅退出信号为SIGTERM，可自定义为其它信号。
    class CMyApplication: public mooon::net::CZookeeperHelper, public mooon::sys::CMainHelper
    {
    public:
        CMyApplication();
    
    private:
        // num_workers 需要启动的CWorker个数
        bool start_workers(
            std::vector<std::thread>* work_threads,
            std::vector<std::shared_ptr<CWorker>>* workers,
            int num_workers);
        void stop_workers(
            std::vector<std::thread>* work_threads,
            std::vector<std::shared_ptr<CWorker>>* workers);
        // 当zookeeper的会话过期后，
        // 需要调用recreate_zookeeper_session重新建立会话
        void recreate_zookeeper_session();
    
    // 实现父类CMainHelper定义的虚拟函数（实为回调函数），
    // 以下五个“on_”函数，均运行在独立的信号线程中，而不是主线程中。
    private:
        // 主线程的调用顺序：
        //    main()
        // -> on_check_parameter() -> on_init()
        // -> on_run() -> on_fini()
        //
        // 注意on_terminated()是由信号触发的，
        // 由独立的信号线程调用，但位于on_init()之后。
        virtual bool on_check_parameter();
        virtual bool on_init(int argc, char* argv[]);
        virtual bool on_run(); // 这里使得配置动态生效
        virtual void on_fini();
        virtual void on_terminated();
    
    // 实现父类CZookeeperHelper定义的虚拟函数（实为回调函数）
    // 以下五个“on_”函数，均运行在独立的zookeeper线程中，而不是主线程中。
    private:
        virtual void on_zookeeper_session_connected(const char* path);
        virtual void on_zookeeper_session_connecting(const char* path);
        virtual void on_zookeeper_session_expired(const char *path);
        virtual void on_zookeeper_session_event(int state, const char *path);
        virtual void on_zookeeper_event(int type, int state, const char *path);
    
    private:
        volatile bool _stop;
        std::mutex _mutex;
        std::condition_variable _cond;
        std::vector<std::thread> _work_threads;
        std::vector<std::shared_ptr<CWorker>> _workers;
    
    private:
        volatile bool _conf_changed; // 配置发生变化
        volatile bool _zookeeper_session_expired; // zookeeper的会话（session）过期
        std::string _zk_nodes; // 存放配置的zookeeper节点列表
        std::string _conf_zkpath; // 配置的zookeeper节点路径
    };
    
    int main(int argc, char* argv[])
    {
        CMyApplication app;
        return mooon::sys::main_template(&app, argc, argv);
    }
    
    static unsigned long long get_current_thread_id()
    {
        std::stringstream ss;
        ss << std::this_thread::get_id();
        return std::stoull(ss.str());
    }
    
    CMyApplication::CMyApplication()
        : _stop(false), _conf_changed(false), _zookeeper_session_expired(false)
    {
        _conf_zkpath = "/tmp/conf";
    }
    
    bool CMyApplication::on_check_parameter()
    {
        // 命令行参数“--zookeeper”不能为空
        return !mooon::argument::zookeeper->value().empty();
    }
    
    bool CMyApplication::on_init(int argc, char* argv[])
    {
        try
        {
            // 以this方式调用的函数，均为CZookeeperHelper提供
            _zk_nodes = mooon::argument::zookeeper->value();
            this->create_session(_zk_nodes);
    
            // zookeeper的会话（session）是异步创建的，
            // 只有连接成功后，方可读取存放在zookeeper上的配置数据。
            for (int i=0; i<5&&!_stop; ++i)
            {
                if (this->is_connected())
                    break;
                else
                    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            }
    
            if (!this->is_connected())
            {
                fprintf(stderr, "Can not connect zookeeper://%s\n", _zk_nodes.c_str());
                return false;
            }
            else
            {
                // 取zookeeper节点数据
                std::string zkdata;
                int n = get_zk_data(_conf_zkpath.c_str(), &zkdata, 4);
                if (n > 4 || zkdata.empty())
                {
                    // 配置数据的大小超出预期
                    fprintf(stderr, "conf size error: %d\n", n);
                    return false;
                }
                else
                {
                    // 如果zkdata不是一个有效的数字，
                    // stoi会抛出异常invalid_argument
                    const int num_workers = std::stoi(zkdata);
    
                    if (num_workers < 1 || num_workers > 10)
                    {
                        fprintf(stderr, "conf error: %d\n", num_workers);
                        return false;
                    }
                    else
                    {
                        return start_workers(&_work_threads, &_workers, num_workers);
                    }
                }
            }
        }
        catch (std::invalid_argument& ex)
        {
            fprintf(stderr, "%s\n", ex.what());
            return false;
        }
        catch (mooon::sys::CSyscallException& ex)
        {
            fprintf(stderr, "%s\n", ex.str().c_str());
            return false;
        }
        catch (mooon::utils::CException& ex)
        {
            fprintf(stderr, "%s\n", ex.str().c_str());
            return false;
        }
    }
    
    bool CMyApplication::on_run()
    {
        while (!_stop)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock); // 等待配置更新或收到退出指令
            if (_stop)
            {
                break;
            }
    
            // 以下实现省略了函数调用抛异常处理
            if (_zookeeper_session_expired)
            {
                // 如果会话过期，则需要重新建会话
                recreate_zookeeper_session();
            }
            if (_stop)
            {
                // 在建立会话过程中，可能收到了停止指令
                break;
            }
            if (_conf_changed)
            {
                _conf_changed = false;
    
                // 读取新的配置
                std::string zkdata;
                int n = get_zk_data(_conf_zkpath.c_str(), &zkdata, 4);
                if (n > 4)
                {
                    // 这种情况下应触发告警
                    // 配置数据的大小超出预期
                    fprintf(stderr, "conf size error: %d\n", n);
                }
                else
                {
                    // 这里可考虑加上优化：
                    // 只有配置确实发生变化时才进行后续操作。
                    const int num_workers = std::stoi(zkdata);
    
                    if (num_workers < 1 || num_workers > 10)
                    {
                        // 这种情况下应触发告警
                        fprintf(stderr, "conf error: %d\n", num_workers);
                    }
                    else
                    {
                        std::vector<std::thread> work_threads;
                        std::vector<std::shared_ptr<CWorker>> workers;
    
                        // 新的配置生效，才停掉原来的，
                        // 防止因为误操破坏配置，导致整个系统崩溃
                        if (!start_workers(&work_threads, &workers, num_workers))
                        {
                            // 这种情况下应触发告警
                        }
                        else
                        {
                            stop_workers(&_work_threads, &_workers);
                            _work_threads.swap(work_threads);
                            _workers.swap(workers);
                        }
                    }
                }
            }
        }
    
        return true;
    }
    
    void CMyApplication::on_fini()
    {
        // 应用退出时被调用
        fprintf(stdout, "Application is about to quit\n");
    }
    
    // 接收到了SIGTERM信号
    void CMyApplication::on_terminated()
    {
        // 一定要最先调用父类CMainHelper的on_terminated
        mooon::sys::CMainHelper::on_terminated();
    
        _stop = true;
        stop_workers(&_work_threads, &_workers);
    
        std::unique_lock<std::mutex> lock(_mutex);
        _cond.notify_one(); // 唤醒等待状态的CMyApplication::run
    }
    
    bool CMyApplication::start_workers(
        std::vector<std::thread>* work_threads,
        std::vector<std::shared_ptr<CWorker>>* workers,
        int num_workers)
    {
        try
        {
            for (int i=0; i<num_workers; ++i)
            {
                std::shared_ptr<CWorker> worker(new CWorker(this, i));
                workers->push_back(worker);
                work_threads->push_back(std::thread(&CWorker::run, worker));
            }
            return true;
        }
        catch(const std::system_error& ex)
        {
            // 如果有部分启动功能应当回退，这里省略了
            fprintf(stderr, "(%d)%s\n", ex.code().value(), ex.what());
            return false;
        }
    }
    
    void CMyApplication::stop_workers(
        std::vector<std::thread>* work_threads,
        std::vector<std::shared_ptr<CWorker>>* workers)
    {
        for (std::vector<std::shared_ptr<CWorker>>::size_type i=0; i<workers->size(); ++i)
        {
            (*workers)[i]->stop();
            if ((*work_threads)[i].joinable())
                (*work_threads)[i].join();
        }
        work_threads->clear();
        workers->clear();
    }
    
    void CMyApplication::recreate_zookeeper_session()
    {
        unsigned int count = 0;
    
        while (!_stop)
        {
            try
            {
                recreate_session();
                _zookeeper_session_expired = false;
                break;
            }
            catch (mooon::utils::CException& ex)
            {
                std::this_thread::sleep_for(std::chrono::milliseconds(2000));
    
                if (0 == count++%30)
                {
                    fprintf(stderr, "recreate zookeeper session failed: (count:%d)%s\n", count, ex.str().c_str());
                }
            }
        }
    }
    
    void CMyApplication::on_zookeeper_session_connected(const char* path)
    {
        fprintf(stdout, "path=%s\n", path);
    }
    
    void CMyApplication::on_zookeeper_session_connecting(const char* path)
    {
        fprintf(stdout, "path=%s\n", path);
    }
    
    void CMyApplication::on_zookeeper_session_expired(const char *path)
    {
        fprintf(stdout, "path=%s\n", path);
    
        std::unique_lock<std::mutex> lock(_mutex);
        _zookeeper_session_expired = true;
        _cond.notify_one(); // 唤醒等待状态的CMyApplication::run
    }
    
    void CMyApplication::on_zookeeper_session_event(int state, const char *path)
    {
        fprintf(stdout, "state=%d, path=%s\n", state, path);
    }
    
    void CMyApplication::on_zookeeper_event(int type, int state, const char *path)
    {
        fprintf(stdout, "type=%d, state=%d, path=%s\n", type, state, path);
    
        if (ZOO_CONNECTED_STATE == state &&
            ZOO_CHANGED_EVENT == type &&
            0 == strcmp(path, _conf_zkpath.c_str()))
        {
            // 配置发生变化
            std::unique_lock<std::mutex> lock(_mutex);
            _conf_changed = true;
            _cond.notify_one(); // 唤醒等待状态的CMyApplication::run
        }
    }
    
    CWorker::CWorker(CMyApplication* app, int index)
        : _app(app), _index(index), _stop(false)
    {
    }
    
    void CWorker::run()
    {
        fprintf(stdout, "Worker[%d/%llu] \033[1;33mstarted\033[m\n", _index, get_current_thread_id());
    
        while (!_stop)
        {
            // 执行具体的业务逻辑操作，这里仅以sleep替代做示范
            std::this_thread::sleep_for(std::chrono::milliseconds(2000));
            fprintf(stdout, "[%s] Worker[\033[1;33m%d\033[m/%llu] is working ...\n",
                mooon::sys::CDatetimeUtils::get_current_time().c_str(),
                _index, get_current_thread_id());
        }
    
        fprintf(stdout, "Worker[%d/%llu] \033[1;33mstopped\033[m\n", _index, get_current_thread_id());
    }

[https_github.com_eyjian_libmooon_blob_master_include_mooon_net_zookeeper_helper.h]: https://github.com/eyjian/libmooon/blob/master/include/mooon/net/zookeeper_helper.h
[https_github.com_eyjian_libmooon_blob_master_include_mooon_sys_main_template.h]: https://github.com/eyjian/libmooon/blob/master/include/mooon/sys/main_template.h