class RandomForest:
    def __init__(self, is_save=False)  # 初始化模型
    def build_train_data(...)        # 构建训练集
    def predict(self, predict_data)  # 执行预测
    @staticmethod
    def word_segmentation(...)       # 分词处理
    @staticmethod
    def pre_treat_data(...)          # 特征预处理

class RandomForestTools:
    @staticmethod
    def train_data_save/clf_load    # 模型保存/加载
    @staticmethod
    def feature_data_save/load       # 特征数据保存/加载
    @staticmethod
    def one_hot_encode_feature(...)  # 特征编码

@staticmethod
def word_segmentation(train_data):
    """中文分词处理"""
    word_segmentation_result = set()
    for word in jieba.lcut(train_data):
        word_segmentation_result.add(word)
    return word_segmentation_result

@staticmethod
def one_hot_encode_feature(data_list, data_set):
    """将文本转换为二进制特征向量"""
    serialize_list = []
    for data in data_list:
        tmp_serialize_list = []
        for key in data_set:
            tmp_serialize_list.append(1 if key in data else 0)
        serialize_list.append(tmp_serialize_list)
    return serialize_list

def build_train_data(self, pre_train_data_list, result_list, train_size=0.9):
    # 1. 分词处理
    train_data_list = [self.word_segmentation(text) for text in pre_train_data_list]
    
    # 2. 特征提取
    train_data_feature = set()
    for data in train_data_list:
        train_data_feature.update(data)
    
    # 3. 特征编码
    data_matrix = self.pre_treat_data(train_data_list, train_data_feature)
    
    # 4. 数据集划分
    data_train, data_test, result_train, result_test = train_test_split(
        data_matrix, result_list, train_size=train_size
    )
    
    # 5. 模型训练
    self.__clf = RandomForestClassifier(n_jobs=-1).fit(data_train, result_train)
    
    # 6. 结果评估
    accuracy = self.__clf.score(data_test, result_test)

# 训练数据
pre_train_data = [
    u'我很开心', u'我非常开心', u'我其实很开心',
    u'我不开心', u'我一点都不开心', u'我很不开心'
]
result_list = [u'开心', u'开心', u'开心', u'不开心', u'不开心', u'不开心']

# 模型训练
rf = RandomForest()
rf.build_train_data(pre_train_data, result_list)

# 执行预测
print(rf.predict(u'你猜我开心吗?'))  # 输出预测结果

# -*- coding: utf-8 -*-

import os
import signal
import threading
import multiprocessing

import redis
from gearman.worker import GearmanWorker, POLL_TIMEOUT_IN_SECONDS

WORKER_PROCESS_PID = '/tmp/multi_gearman_worker.pid'


class MultiGearmanWorker(GearmanWorker):
    """多进程Gearman worker"""
    def __init__(self, host_list=None, redis_host=None, redis_port=None, pid=WORKER_PROCESS_PID):
        super(MultiGearmanWorker, self).__init__(host_list=host_list)
        self.redis_host = redis_host
        self.redis_port = redis_port
        self.pid = pid

    def work(self, poll_timeout=POLL_TIMEOUT_IN_SECONDS, process=multiprocessing.cpu_count()):
        """开始作业，进程阻塞
        :param poll_timeout: gearman连接超时时间，值越小worker召回越快但请求越频繁
        :param process: 工作进程数，默认为CPU核心数
        :return:
        """
        print('Clear last process.')
        self.gearman_worker_exit()
        print('Ready to start %d process for work.' % process)
        gm_poll = multiprocessing.Pool(process)
        for x in range(0, process):
            gm_poll.apply_async(gearman_work, (self, poll_timeout, self.pid))
        gm_poll.close()
        gm_poll.join()
        # 正常退出则删除子进程PID文件
        if os.path.isfile(self.pid):
            os.remove(self.pid)

        print('Multi gearman worker exit.')

    def gearman_worker_exit(self):
        """结束子进程"""
        if not os.path.isfile(self.pid):
            return True

        with open(self.pid, 'r+') as f:
            for pid in f.readlines():
                pid = int(pid)
                try:
                    os.kill(pid, signal.SIGKILL)
                    print('Kill process %d.' % pid)
                except OSError:
                    print('Process %d not exists' % pid)
                    continue
        os.remove(self.pid)
        print('Remove process pid file.')
        return True

# 子进程使用的Gearman工作开关标识
GEARMAN_CONTINUE_WORK = True


def gearman_work(gm_worker, poll_timeout=POLL_TIMEOUT_IN_SECONDS, pid=WORKER_PROCESS_PID):
    """以多进程方式开启Gearman worker"""
    try:
        # 记录子进程PID以便主进程重启后清除未退出的子进程
        with open(pid, 'a+') as f:
            f.write("%d%s" % (os.getpid(), os.linesep))

        print('Child process start for work.')
        continue_working = True
        worker_connections = []
        d = threading.Thread(name='monitor', target=gearman_monitor,
                             args=(gm_worker.redis_host, gm_worker.redis_port))
        d.start()

        def continue_while_connections_alive(any_activity):
            return gm_worker.after_poll(any_activity)

        # 轮询连接，等待任务
        while continue_working and GEARMAN_CONTINUE_WORK:
            worker_connections = gm_worker.establish_worker_connections()
            continue_working = gm_worker.poll_connections_until_stopped(
                worker_connections, continue_while_connections_alive, timeout=poll_timeout)

        # 关闭所有连接
        for current_connection in worker_connections:
            current_connection.close()

        print('Gearman worker closed')
        return None
    except Exception as e:
        print(e)


def gearman_monitor(redis_host, redis_port):
    """监听Redis的发布订阅信号"""
    global GEARMAN_CONTINUE_WORK
    print('Start gearman monitor.')
    while GEARMAN_CONTINUE_WORK:
        try:
            sub = redis.StrictRedis(redis_host, redis_port).pubsub()
            sub.subscribe('hot')
            for i in sub.listen():
                if isinstance(i.get('data'), str):
                    if i.get('data') == 'exit':
                        print('Gearman monitor receive restart signal.')
                        GEARMAN_CONTINUE_WORK = False
                        sub.unsubscribe('hot')
                        break
        except Exception as e:
            print(e)
            try:
                sub.unsubscribe('hot')
            except Exception:
                pass

    print('Gearman monitor closed')


if __name__ == '__main__':
    def test_multi_gearman_worker(worker, job):
        print(worker)
        print(job)

    # 初始化多进程Gearman worker
    gearman_worker = MultiGearmanWorker(
        host_list=('127.0.0.1:4730',), 
        redis_host='127.0.0.1', 
        redis_port=6379
    )
    # 注册任务
    gearman_worker.register_task('test_multi_gearman_worker', test_multi_gearman_worker)
    # 启动工作
    gearman_worker.work()