队列queue logging mysql连接池 多线程 redis集群 countdownlatch python数组按照对象字段排序 追加写文件 格式化时间

python-3.8 
pip3 install DBUtils==2.0.2  
pip3 install redis-py-cluster==2.1.3 
如遇pip3安装失败则解压：redis-py-cluster-2.1.3.tar.gz 并执行 
python setup.py build  
python setup.py install
TIPS:不要纠结版本pip在线安装的版本依赖等等，如遇问题，离线安装即可。
redis-py-cluster-2.1.3.tar.gz : https://github.com/Grokzen/redis-py-cluster/releases
DBUtils-2.0.2.tar.gz : https://files.pythonhosted.org/packages/cb/71/c931d8ed80f785f8d5c690ca0e8e6d512159581685551a4adbf94db94d2e/DBUtils-2.0.2.tar.gz

# -*- coding: UTF-8 -*-
# !/usr/bin/python3import json
import logging
import math
import queue
import threading
import timeimport pymysql
from dbutils.pooled_db import PooledDB
from rediscluster import RedisCluster###############参数配置######################
host = "127.0.0.1"
port = 3306
db = "dbname"
user = "user"
password = "password"
# 连接池最小数
minCached = 1
# 连接池最大数
maxCached = 20
# 并发执行线程数
threads = 1
# 睡眠时间-秒
sleepSecond = 0
# mysql和redis中使用流量差值
diffFlowMb = 1
# redis 集群地址
startup_nodes = [{"host": "127.0.0.1", "port": "7001"},{"host": "127.0.0.1", "port": "7002"},{"host": "127.0.0.1", "port": "7003"},{"host": "127.0.0.1", "port": "7004"},{"host": "127.0.0.1", "port": "7005"},{"host": "127.0.0.1", "port": "7006"}]
# 分页查询数量
pageSize = 1000
# 流量重置开始时间
lastUpdateStartTime = "2021-10-18 13:30:00"
# 流量重置结束时间
lastUpdateEndTime = "2021-10-20 18:30:00"############################################
redisConn = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)job_begin = time.time() * 1000
producerFlag = False
logging.basicConfig(filename="log.log", level=logging.INFO,format="%(asctime)s: %(thread)d - %(levelname)s - %(message)s",datefmt="%Y-%m-%d %H:%M:%S")class CountDownLatch:def __init__(self, count):self.count = countself.condition = threading.Condition()def awaits(self):try:self.condition.acquire()while self.count > 0:self.condition.wait()finally:self.condition.release()def countDown(self):try:self.condition.acquire()self.count -= 1self.condition.notifyAll()finally:self.condition.release()class Mysql(object):"""MYSQL数据库对象，负责产生数据库连接 , 此类中的连接采用连接池实现获取连接对象：conn = Mysql.getConn()释放连接对象;conn.close()或del conn"""# 连接池对象__pool = Nonedef __init__(self):# 数据库构造函数，从连接池中取出连接，并生成操作游标self._conn = Mysql.__getConn()self._cursor = self._conn.cursor()@staticmethoddef __getConn():"""@summary: 静态方法，从连接池中取出连接@return MySQLdb.connection"""if Mysql.__pool is None:__pool = PooledDB(creator=pymysql, mincached=minCached, maxcached=maxCached,host=host, port=port, user=user, passwd=password,db=db, use_unicode=False, charset="utf8",cursorclass=pymysql.cursors.DictCursor)return __pool.connection()def getAll(self, sql, param=None):"""@summary: 执行查询，并取出所有结果集@param sql:查询ＳＱＬ，如果有查询条件，请只指定条件列表，并将条件值使用参数[param]传递进来@param param: 可选参数，条件列表值（元组/列表）@return: result list(字典对象)/boolean 查询到的结果集"""if param is None:count = self._cursor.execute(sql)else:count = self._cursor.execute(sql, param)if count > 0:result = self._cursor.fetchall()else:result = Falsereturn resultdef dispose(self, isEnd=1):"""@summary: 释放连接池资源"""if isEnd == 1:self.end('commit')else:self.end('rollback');self._cursor.close()self._conn.close()class PackageDiff:def __init__(self, id, imsi, initFlow, mysqlUseFlow, redisUseFlow, diff, packageStatus, updateType,lastUpdateTime):self.id = idself.imsi = imsiself.initFlow = initFlowself.mysqlUseFlow = mysqlUseFlowself.redisUseFlow = redisUseFlowself.diff = diffself.packageStatus = packageStatusself.updateType = updateTypeself.lastUpdateTime = lastUpdateTimedef __repr__(self):return repr((self.id, self.imsi, self.initFlow, self.mysqlUseFlow,self.redisUseFlow, self.diff, self.packageStatus, self.updateType, self.lastUpdateTime))def toCsv(self):return self.id + "," + self.imsi + "," + self.initFlow \+ "," + self.mysqlUseFlow + "," + self.redisUseFlow + "," + self.diff + "," + self.packageStatus \+ "," + self.updateType + "," + self.lastUpdateTimedef producer(threadNum):global producerFlagstart = time.time() * 1000count = 0pageIndex = 0sql = "select a.id,a.imsi,a.init_flow,a.total_use_flow,a.init_flow,a.package_status,a.update_type," \"a.last_update_time from t_css_vsim_packages a join " \"(select id from t_css_vsim_packages where last_update_time > %s and last_update_time < %s limit %s," \"%s) b on a.id = b.id; "try:packages = mysql.getAll(sql, [lastUpdateStartTime, lastUpdateEndTime, pageIndex, pageSize])while packages:for col in packages:packageQueue.put(col)count += 1pageIndex += pageSizepackages = mysql.getAll(sql, [lastUpdateStartTime, lastUpdateEndTime, pageIndex, pageSize])producerFlag = Trueexcept Exception as e:logging.error("fetch all packages record failed exception: ", e)logging.info("[thread-producer" + str(threadNum) + "]producer queue is complete . count : %s . cost: %s ms ",str(count),str(time.time() * 1000 - start))def consumer(threadNum):start = time.time() * 1000while True:if producerFlag and packageQueue.empty():logging.info("[thread-consumer-%s] producer complete and queue is empty. cost: %s ms", str(threadNum),str(time.time() * 1000 - start))countDownLatch.countDown()returnelse:try:package = packageQueue.get(timeout=10)logging.info("[thread-consumer-%s] one record: %s.left size: %s", str(threadNum), str(package),str(packageQueue.qsize()))checkAndExport(package)time.sleep(sleepSecond)except Exception as e:logging.warning("[thread-consumer-%s] queue is empty . left size : %s", packageQueue.qsize(), e)def byteToStr(param):return str(param, encoding='utf-8')def dateToStr(param):return param.strftime("%Y-%m-%d %H:%M:%S")def checkAndExport(package):packageId = byteToStr(package['id'])mysqlUseFlow = package['total_use_flow']vsimPackage = redisConn.hget("CACHE_PACKAGE", packageId)if vsimPackage is None:logging.warning("redis not found. packageId: %s imsi: %s", packageId, str(package['imsi']))else:vsimPackageObj = json.loads(vsimPackage)redisUseFlow = vsimPackageObj['totalUseFlow']diff = int(math.fabs(redisUseFlow - mysqlUseFlow))if diff > (diffFlowMb * 1024 * 1024):result.append(PackageDiff(packageId, str(package['imsi']), str(package['init_flow']), str(mysqlUseFlow),str(redisUseFlow), str(diff),str(package['package_status']), str(package['update_type']),dateToStr(package['last_update_time'])))def sortAndWrite():start = time.time() * 1000logging.info("[main] ready to write file. size: %s", len(result))sortResult = sorted(result, key=lambda package: package.diff, reverse=False)file = open("result.csv", "a")file.write("packageId,imsi,initFlow,mysqlUseFlow,redisUseFlow,diff,""packageStatus,updateType,lastUpdateTime\n")for info in sortResult:file.write(info.toCsv() + '\n')file.close()logging.info("[main] write file complete. cost: %s ms.", str(time.time() * 1000 - start))if __name__ == "__main__":packageQueue = queue.Queue()mysql = Mysql()result = []countDownLatch = CountDownLatch(threads)try:threading.Thread(target=producer, args=(1,)).start()for i in range(threads):threading.Thread(target=consumer, args=(i,)).start()countDownLatch.awaits()sortAndWrite()except Exception as e:logging.error("[main] thread raise exception: ", e)finally:logging.info("[main] job execute successful. total cost: %s ms", str(time.time() * 1000 - job_begin))