# -*- coding:utf-8 -*- # 配置文件初始化 # 主要是一些 db 的初始化创建 # 简单 Class 类函数的 初始化 import asyncio import json import logging import os from concurrent.futures import ThreadPoolExecutor from configparser import ConfigParser from logging.handlers import TimedRotatingFileHandler import aredis import redis import requests import uvloop from aiokafka import AIOKafkaProducer from elasticsearch import Elasticsearch from requests import adapters adapters.DEFAULT_POOLSIZE = 100000 asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) class CfgBaseInit(object): """ config: 配置文件对象 project_name: 项目名称 start_mode: 项目启动模式 """ config = ConfigParser() config.read(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.ini")) project_name = config.get("CONFIG", "project_name") project_path = os.path.join(os.path.abspath(os.path.dirname(__file__)).split(project_name)[0], project_name) start_model = config.get("CONFIG", "start_model").upper() __authentication = config.get("CONFIG", "authentication") authentication = json.loads(__authentication if "'" not in __authentication else __authentication.replace("'", '"')) __headers = config.get("CONFIG", "headers") headers = json.loads(__headers if "'" not in __headers else __headers.replace("'", '"')) bind = config.get(section=f"PROJECT-{start_model}", option="host") _executor = ThreadPoolExecutor(max_workers=50000) # 线程池 _async_kafka_session = None kafka_server = config.get(f"BEHAVE-{start_model}", "kafka_server") @staticmethod def get_my_event_loop(): """ 获取当前 loop 循环事件管理对象 :return: """ event_loop = asyncio.get_event_loop() # asyncio.set_event_loop(loop=event_loop) return event_loop @classmethod def executor_submit(cls, *args, **kwargs): """ 改写 executor.submit 方法，加入默认的异常捕获功能； 使用方式与 原来的 executor.submit 相同 :param args: :param kwargs: :return: """ if asyncio.iscoroutinefunction(args[0]): loop = asyncio.new_event_loop() if not loop.is_running(): executor_future = loop.run_until_complete(args[0](args[1:])) # 处理 future 函数 or 对象 return executor_future else: thread_name_prefix = kwargs.get("thread_name_prefix", "") if not thread_name_prefix: thread_name_prefix = args[0].__name__ if "thread_name_prefix" in kwargs: del kwargs['thread_name_prefix'] cls._executor._thread_name_prefix = thread_name_prefix executor_future = cls._executor.submit(*args, **kwargs) # 处理 普通 function； # 线程任务增加任务名称前缀，与线程名称相同. # 默认为 function.__name__ executor_future.__setattr__("name", thread_name_prefix) executor_future.add_done_callback(cls._executor_callback) return executor_future @classmethod def _executor_callback(cls, worker): """ 任务池中任务异常捕获回调函数 抛出异常 :param worker: 任务 :return: """ worker_exception = worker.exception() if worker_exception: logger.exception("Worker return exception: {}".format(worker_exception)) raise worker_exception class Logger(object): # 日志 log_path = CfgBaseInit.config.get("LOG", "log_path") when = CfgBaseInit.config.get("LOG", "when") backupCount = int(CfgBaseInit.config.get("LOG", "backupCount")) formatter = logging.Formatter( '%(asctime)s %(name)s %(process)d %(thread)d %(threadName)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') handler = TimedRotatingFileHandler(log_path, when=when, backupCount=backupCount) handler.setFormatter(formatter) logger = logging.getLogger(CfgBaseInit.project_name) logger.addHandler(handler) logger.setLevel(logging.DEBUG) def __init__(self, cls): self._cls = cls def __call__(self, *args, **kwargs): if not hasattr(self._cls, 'logger'): self._cls.logger = logger return self._cls(*args, **kwargs) @staticmethod def replace_blank(msg, *args, **kwargs): msg = str(msg) return f"{msg % args}".replace("\r", " ").replace("\n", " ") def debug(self, msg, *args, **kwargs): self.logger.debug(self.replace_blank(msg, *args, **kwargs)) def info(self, msg, *args, **kwargs): self.logger.info(self.replace_blank(msg, *args, **kwargs)) def warning(self, msg, *args, **kwargs): self.logger.warning(self.replace_blank(msg, *args, **kwargs)) def error(self, msg, *args, **kwargs): self.logger.error(self.replace_blank(msg, *args, **kwargs)) def exception(self, msg, *args, **kwargs): self.logger.exception(self.replace_blank(msg, *args, **kwargs)) def critical(self, msg, *args, **kwargs): self.logger.critical(self.replace_blank(msg, *args, **kwargs)) logger = Logger(Logger) class RedisInit(CfgBaseInit): """ Redis 对象初始化 """ section = "REDIS-{}".format(CfgBaseInit.start_model.upper()) host = CfgBaseInit.config.get(section=section, option="host") port = CfgBaseInit.config.get(section=section, option="port") db = CfgBaseInit.config.get(section=section, option="db") host_bak = CfgBaseInit.config.get(section=section, option="host_bak") port_bak = CfgBaseInit.config.get(section=section, option="port_bak") db_bak = CfgBaseInit.config.get(section=section, option="db_bak") Pool = redis.ConnectionPool(host=host, port=port, db=db, max_connections=None, decode_responses=True, socket_keepalive=False) Pool_bak = redis.ConnectionPool(host=host_bak, port=port_bak, db=db_bak, max_connections=None, decode_responses=True) conn = redis.Redis(connection_pool=Pool) class AsyncRedisInit(CfgBaseInit): """ redis 异步连接池初始化 pipeline_demo： async def pipeline_test(): conn = aredis.StrictRedis(connection_pool=AsyncRedisInit.Pool) p = await conn.pipeline() await p.hgetall("process_cache\0015f371f42f286b2eb461e75a437162c6e@cccxx@abrs_reco@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@0@0_0") a = await p.execute() return a client_demo: async def conn_test(): conn = aredis.StrictRedis(connection_pool=AsyncRedisInit.Pool) a = await conn.hgetall("process_cache\0015f371f42f286b2eb461e75a437162c6e@cccxx@abrs_reco@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@0@0_0") return a """ section = "REDIS-{}".format(CfgBaseInit.start_model.upper()) host = CfgBaseInit.config.get(section=section, option="host") port = CfgBaseInit.config.get(section=section, option="port") try: password = CfgBaseInit.config.get(section=section, option="password") except Exception: password = None db = CfgBaseInit.config.get(section=section, option="db") host_bak = CfgBaseInit.config.get(section=section, option="host_bak") port_bak = CfgBaseInit.config.get(section=section, option="port_bak") try: password_bak = password except Exception: password_bak = None db_bak = CfgBaseInit.config.get(section=section, option="db_bak") # todo 不能使用链接池，链接池不能自动释放 conn Pool = aredis.ConnectionPool(host=host, port=port, db=db, password=password, max_connections=None, decode_responses=True, socket_keepalive=False) Pool_bak = aredis.ConnectionPool(host=host_bak, port=port_bak, db=db_bak, password=password_bak, max_connections=None, decode_responses=True, socket_keepalive=False) conn = aredis.StrictRedis(connection_pool=Pool) class ElasticSearchInit(CfgBaseInit): """ ElasticSearch 对象初始化 """ section = "ES-{}".format(CfgBaseInit.start_model.upper()) index = CfgBaseInit.config.get(section=section, option="index") fc_index = CfgBaseInit.config.get(section=section, option="fc_index") try: es_timeout = float(CfgBaseInit.config.get(section=section, option="timeout")) except Exception as e: es_timeout = 10 es_client = Elasticsearch(hosts=[CfgBaseInit.config.get(section=section, option="host")], timeout=es_timeout) fc_es_client = Elasticsearch(hosts=CfgBaseInit.config.get(section=section, option="fc_host").split(","), timeout=es_timeout) class BehaviorAPI(CfgBaseInit): """ 行为数据对接接口 Behavior Api 初始化 """ section = "BEHAVIOR_API-{}".format(CfgBaseInit.start_model.upper()) url = CfgBaseInit.config.get(section=section, option="url") @staticmethod def push_behavior_log(data): data = json.dumps(data) r = requests.post(BehaviorAPI.url, data=data, headers=CfgBaseInit.headers) return r class AsyncKafka(CfgBaseInit): """异步请求模块 """ _async_kafka_session = None @classmethod async def __set_async_session(cls): producer = AIOKafkaProducer(loop=cls.get_my_event_loop(), value_serializer=lambda v: json.dumps(v).encode('utf-8'), bootstrap_servers=CfgBaseInit.kafka_server) await producer.start() cls._async_kafka_session = producer return producer @classmethod async def async_push(cls, payload): """异步请求接口 """ request_session = cls._async_kafka_session or await cls.__set_async_session() print(request_session) await request_session.send_and_wait("eip_rec_behave", payload) if __name__ == '__main__': a = BehaviorAPI.push_behavior_log({"a": 1}) print(a) print(a.content)