一个简单的python 多进程多线程处理redis或kafka数据脚本
·
背景:当前需要对redis或者kafka数据做相关清洗后，发布到redis或kafka中，其中利用多进程多线程方式进行处理。下面完整介绍相关脚本。
配置文件：config.yaml
# Version: 5.0.0
#取数据脚本
get_msg:
thread_num: 8
#数据分发处理
worker:
max_process: 8
thread_num: 256
fcgi:
get_api_url: http://127.0.0.1/ww/ss/dd_de3
#kafka连接信息
kafka:
#填写主机名
host:
- db1:9092
- db2:9092
#数据源topic
topic_in: getMessage
#取清洗后的数据的topic
topic_out: monitorMessage
#入库数据流
topic_srvdb: resultMessage
#web数据库
mysql_web:
host: 127.0.0.1
port: 3306
user: king_admin
password: Kindg#283&Ghwws
database: uswere
#web ip，有端口时需要加上端口
web_ip: 127.0.0.1
#主消息队列redis
redis_mq:
unix_socket_path: '/dev/shm/redis_mq.sock'
host: 127.0.0.1
port: 6380
password: kwi4544RDf
database: 0
#车辆数据
queue_key_name: _jobs
#高速缓存redis
redis_cache:
unix_socket_path: '/dev/shm/redis_cache.sock'
host: 127.0.0.1
port: 6383
password: kwi4544RDf
database: 0
queue_key_name:
#映射缓存服务
web_tools:
listen: 0.0.0.0:9002
守护进程文件 daemon.py
python2 版本：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import atexit
from signal import SIGTERM ,SIGKILL
import os,sys,time
class Daemon:
"""
A generic daemon class.
Usage: subclass the Daemon class and override the run() method
"""
def __init__(self, pidfile, stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
self.stdin = stdin
#self.stdout = stdout
self.stderr = stderr
self.pidfile = pidfile
def daemonize(self):
"""
do the UNIX double-fork magic, see Stevens' "Advanced
Programming in the UNIX Environment" for details (ISBN 0201563177)
http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16
"""
try:
pid = os.fork()
if pid > 0:
# exit first parent
sys.exit(0)
except OSError, e:
sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
sys.exit(1)
# decouple from parent environment
os.chdir(os.path.dirname(os.path.abspath(__file__)))
os.setsid()
os.umask(0)
# do second fork
try:
pid = os.fork()
if pid > 0:
# exit from second parent
sys.exit(0)
except OSError, e:
sys.stderr.write("fork #2 failed: %d (%s)\n" % (e.errno, e.strerror))
sys.exit(1)
# redirect standard file descriptors
sys.stdout.flush()
sys.stderr.flush()
si = file(self.stdin, 'r')
#so = file(self.stdout, 'a+')
se = file(self.stderr, 'a+', 0)
os.dup2(si.fileno(), sys.stdin.fileno())
#os.dup2(so.fileno(), sys.stdout.fileno())
os.dup2(se.fileno(), sys.stderr.fileno())
# write pidfile
atexit.register(self.delpid)
pid = str(os.getpid())
file(self.pidfile,'w+').write("%s\n" % pid)
def delpid(self):
os.remove(self.pidfile)
def kill_child_processes(self,parent_pid, sig=SIGKILL):
try:
try:
pgid = os.getpgid(int(parent_pid))
#print 'killpg',pgid
os.killpg(pgid,sig)
except Exception, e:
ps_command = subprocess.Popen("ps -o pid --ppid %d --noheaders" % parent_pid, shell=True, stdout=subprocess.PIPE)
ps_output = ps_command.stdout.read()
retcode = ps_command.wait()
assert retcode == 0, "ps command returned %d" % retcode
for pid_str in ps_output.split("\n")[:-1]:
os.kill(int(pid_str), sig)
except AssertionError:
return
except Exception, e:
return
def check_pid(self,pid):
""" Check For the existence of a unix pid. """
try:
os.kill(pid, 0)
except OSError:
return False
else:
return True
def start(self):
"""
Start the daemon
"""
# Check for a pidfile to see if the daemon already runs
try:
pf = file(self.pidfile,'r')
pid = int(pf.read().strip())
pf.close()
except IOError:
pid = None
if pid and self.check_pid(pid):
message = "pidfile %s already exist. Daemon already running?\n"
sys.stderr.write(message % self.pidfile)
sys.exit(1)
# Start the daemon
self.daemonize()
self.run()
def stop(self):
"""
Stop the daemon
"""
# Get the pid from the pidfile
try:
pf = file(self.pidfile,'r')
pid = int(pf.read().strip())
pf.close()
except IOError:
pid = None
if not pid:
message = "pidfile %s does not exist. Daemon not running?\n"
sys.stderr.write(message % self.pidfile)
return # not an error in a restart
# Try killing the daemon process
try:
while 1:
self.kill_child_processes(pid, SIGTERM)
os.kill(pid, SIGTERM)
time.sleep(0.1)
except OSError, err:
err = str(err)
if err.find("No such process") > 0:
if os.path.exists(self.pidfile):
os.remove(self.pidfile)
else:
print str(err)
sys.exit(1)
def restart(self):
"""
Restart the daemon
"""
self.stop()
self.start()
def run(self):
"""
You should override this method when you subclass Daemon. It will be called after the process has been
daemonized by start() or restart().
"""
daemon.py python3 版本：
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import atexit
import subprocess
from signal import SIGTERM, SIGKILL
import os
import sys
import time
class Daemon:
"""
A generic daemon class for Python 3.
Usage: subclass the Daemon class and override the run() method
"""
def __init__(self, pidfile, stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
self.stdin = stdin
self.stdout = stdout # 恢复stdout属性
self.stderr = stderr
self.pidfile = pidfile
def daemonize(self):
"""
执行UNIX双叉创建守护进程，参考Stevens的《UNIX环境高级编程》
"""
try:
pid = os.fork()
if pid > 0:
# 退出第一个父进程
sys.exit(0)
except OSError as e:
sys.stderr.write(f"fork #1 failed: {e.errno} ({e.strerror})\n")
sys.exit(1)
# 脱离父进程环境
os.chdir(os.path.dirname(os.path.abspath(__file__)))
os.setsid()
os.umask(0)
# 第二次fork
try:
pid = os.fork()
if pid > 0:
# 退出第二个父进程
sys.exit(0)
except OSError as e:
sys.stderr.write(f"fork #2 failed: {e.errno} ({e.strerror})\n")
sys.exit(1)
# 重定向标准文件描述符
sys.stdout.flush()
sys.stderr.flush()
# Python 3 使用 open() 替代 file()
with open(self.stdin, 'r') as si, \
open(self.stdout, 'a+') as so, \
open(self.stderr, 'a+', buffering=0) as se:
os.dup2(si.fileno(), sys.stdin.fileno())
os.dup2(so.fileno(), sys.stdout.fileno())
os.dup2(se.fileno(), sys.stderr.fileno())
# 注册退出时删除pid文件
atexit.register(self.delpid)
pid = str(os.getpid())
with open(self.pidfile, 'w+') as f:
f.write(f"{pid}\n")
def delpid(self):
"""删除pid文件"""
if os.path.exists(self.pidfile):
os.remove(self.pidfile)
def kill_child_processes(self, parent_pid, sig=SIGKILL):
"""终止指定父进程的所有子进程"""
try:
try:
pgid = os.getpgid(int(parent_pid))
os.killpg(pgid, sig)
except Exception:
# 使用subprocess获取子进程PID并终止
ps_command = subprocess.Popen(
f"ps -o pid --ppid {parent_pid} --noheaders",
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True # Python 3 推荐使用text替代universal_newlines
)
ps_output, ps_error = ps_command.communicate()
retcode = ps_command.returncode
if retcode != 0:
return
for pid_str in ps_output.strip().split('\n'):
if pid_str.strip():
os.kill(int(pid_str.strip()), sig)
except Exception:
return
def check_pid(self, pid):
"""检查指定PID的进程是否存在"""
try:
os.kill(pid, 0)
except OSError:
return False
else:
return True
def start(self):
"""启动守护进程"""
# 检查pid文件是否存在，判断进程是否已运行
pid = None
try:
with open(self.pidfile, 'r') as pf:
pid = int(pf.read().strip())
except IOError:
pass
if pid and self.check_pid(pid):
message = f"pidfile {self.pidfile} already exist. Daemon already running?\n"
sys.stderr.write(message)
sys.exit(1)
# 启动守护进程
self.daemonize()
self.run()
def stop(self):
"""停止守护进程"""
# 从pid文件读取PID
pid = None
try:
with open(self.pidfile, 'r') as pf:
pid = int(pf.read().strip())
except IOError:
pass
if not pid:
message = f"pidfile {self.pidfile} does not exist. Daemon not running?\n"
sys.stderr.write(message)
return
# 尝试终止守护进程
try:
while True:
self.kill_child_processes(pid, SIGTERM)
os.kill(pid, SIGTERM)
time.sleep(0.1)
except OSError as err:
err_str = str(err)
if "No such process" in err_str:
if os.path.exists(self.pidfile):
os.remove(self.pidfile)
else:
print(err_str)
sys.exit(1)
def restart(self):
"""重启守护进程"""
self.stop()
self.start()
def run(self):
"""
子类需要重写此方法，该方法会在daemonize后被调用
在此方法中实现守护进程的核心业务逻辑
"""
raise NotImplementedError("Subclasses must override the run() method")
# 测试示例（可选）
if __name__ == "__main__":
import argparse
# 示例子类
class MyDaemon(Daemon):
def run(self):
# 示例：每秒写入一次日志
while True:
with open('/tmp/daemon_test.log', 'a') as f:
f.write(f"Daemon running at {time.ctime()}\n")
time.sleep(1)
# 命令行参数解析
parser = argparse.ArgumentParser(description='Python 3 Daemon Example')
parser.add_argument('action', choices=['start', 'stop', 'restart'])
args = parser.parse_args()
# 初始化守护进程
daemon = MyDaemon('/tmp/daemon.pid', stdout='/tmp/daemon.out', stderr='/tmp/daemon.err')
# 执行对应操作
if args.action == 'start':
daemon.start()
elif args.action == 'stop':
daemon.stop()
elif args.action == 'restart':
daemon.restart()
下面代码基本上都是用python2 运行 如果需要运行python3 请自行适配：
抽取redis数据到redis_mq模块 脚本
#!/usr/bin/python
#-*- coding: utf-8 -*-
# 抽取redis数据到redis_mq模块
# 作者：树下水月
import redis
import requests
import json
import os
import re
import sys
import struct
import random
import yaml,logging
from logging.handlers import TimedRotatingFileHandler,RotatingFileHandler
import time,timeit,datetime,sys,os,threading
from daemon import Daemon
import traceback
import StringIO
import pickle
import pylru
import pickle
import MySQLdb
import base64
import hashlib
import Queue
reload(sys)
sys.setdefaultencoding('utf-8')
message_queue = Queue.Queue(2000)
class MyDaemon(Daemon):
def execute_sql(self,sql,action='select'):
try:
if self.db is None:
self.db = MySQLdb.connect(host=self.config['mysql_web']['host'],port=self.config['mysql_web']['port'], user=self.config['mysql_web']['user'], passwd=self.config['mysql_web']['password'], db=self.config['mysql_web']['database'],charset="utf8")
try:
self.db.ping()
except MySQLdb.Error,e:
self.db = MySQLdb.connect(host=self.config['mysql_web']['host'],port=self.config['mysql_web']['port'], user=self.config['mysql_web']['user'], passwd=self.config['mysql_web']['password'], db=self.config['mysql_web']['database'],charset="utf8")
mysql_web = self.db.cursor(MySQLdb.cursors.DictCursor)
r = mysql_web.execute(sql)
if action=='select':
r = mysql_web.fetchall()
elif action=='update':
pass
elif action=='insert':
r = self.db.insert_id()
mysql_web.close()
return r
except Exception, e:
logging.exception('连接数据库时错误: %s', str(e))
r = None
if action=='select':
r = []
elif action=='update':
pass
elif action=='insert':
r = 0
return r
def load_cache(self):
self.device_to_device = {}#第三方设备ID<=>设备ID映射
if os.path.isfile(os.path.dirname(os.path.abspath(__file__))+"/data/device_dict.dat"):
with open(os.path.dirname(os.path.abspath(__file__))+"/data/device_dict.dat") as f:
d_str = f.read()
self.device_to_device = pickle.loads(d_str)
self.location_to_location = {}#第三方卡口ID<=>卡口ID映射
if os.path.isfile(os.path.dirname(os.path.abspath(__file__))+"/data/location_dict.dat"):
with open(os.path.dirname(os.path.abspath(__file__))+"/data/location_dict.dat") as f:
d_str = f.read()
self.location_to_location = pickle.loads(d_str)
def format_msg(self,msg):
# {u'VehicleRearItem': u'99', u'TabID': None, u'SafetyBelt': 1, u'DescOfRearItem': None, u'VehicleLength': None, u'NameOfPassedRoad': None, u'BrandReliability': u'0', u'PlateNo': u'\u9c81DQD171', u'VehicleShielding': None, u'VehicleColorDepth': u'0', u'RightBtmX': 12, u'RightBtmY': 12, u'HasPlate': u'1', u'Direction': u'9', u'VehicleClass': None, u'PlateClass': u'99', u'VehicleRoof': None, u'SourceID': u'37040020195034141244022019090621002016892', u'VehicleWindow': None, u'TollgateID': u'37040000001210000000', u'DrivingStatusCode': None, u'WheelPrintedPattern': None, u'PlateNoAttach': None, u'VehicleWidth': None, u'VehicleBrand': u'0', u'PlateCharReliability': None, u'IsAltered': u'0', u'Sunvisor': 0, u'PassTime': u'20190906210020', u'VehicleWheel': None, u'SideOfVehicle': None, u'LaneNo': 1, u'FeatureList': None, u'VehicleFrontItem': u'99', u'VehicleHood': None, u'FilmColor': u'3', u'UsingPropertiesCode': None, u'VehicleTrunk': None, u'IsModified': u'0', u'StorageUrl1': u'http://37.79.2.176:8092/api/getImageData.jpg?picUrl=ZnRwOi8vaHRmdHA6aHVpdG9uZ0AzNy43OS4yLjE3ODoyMS8yMDE5MDkvaGlrLzYzODE3MTAxOTAwMC9ray8wNi8yMS8yMDE5MDkwNjIxMDAwMzgzN182MzgxNzEwMTkwMDBfMDFfMV80MzkxNjEuanBn&collectiondate=20190906', u'StorageUrl3': None, u'StorageUrl2': None, u'StorageUrl5': None, u'StorageUrl4': None, u'Calling': 0, u'VehicleBodyDesc': None, u'NumOfPassenger': None, u'VehicleHeight': None, u'DeviceID': u'37040000001210000000', u'NationalityCode': None, u'SubImageList': {u'SubImageInfoObject': [{u'StoragePath': u'http://37.79.2.176:8092/api/getImageData.jpg?picUrl=ZnRwOi8vaHRmdHA6aHVpdG9uZ0AzNy43OS4yLjE3ODoyMS8yMDE5MDkvaGlrLzYzODE3MTAxOTAwMC9ray8wNi8yMS8yMDE5MDkwNjIxMDAwMzgzN182MzgxNzEwMTkwMDBfMDFfMV80MzkxNjEuanBn&collectiondate=20190906', u'Data': None, u'FeatureInfoObject': None, u'ImageID': None, u'Width': 1, u'Height': 1, u'DeviceID': None, u'ShotTime': u'20190906210020', u'Type': u'01', u'FileFormat': u'Jpeg', u'EventSort': None}]}, u'PlateReliability': u'0', u'RearviewMirror': None, u'VehicleDoor': None, u'HitMarkInfo': u'0', u'LeftTopX': 12, u'LeftTopY': 12, u'InfoKind': 1, u'VehicleChassis': None, u'CarOfVehicle': None, u'VehicleModel': None, u'PlateColor': u'99', u'MotorVehicleID': u'370400201950341412440220190906210020168920200000', u'PlateDescribe': None, u'IsCovered': u'0', u'IsDecked': u'0', u'VehicleColor': u'99', u'DescOfFrontItem': None, u'VehicleStyles': None, u'Speed': None}
try:
row = {}
insert_id = 0 #初始化写入标识
#print msg
if msg['PlateNo']:
plate = msg['PlateNo']
if len(plate) < 7 or plate=='未识别' or plate=='无' or plate=='无车牌' or plate=='Failure' or plate=="EmptyPlate":
row['license_plate'] = '无牌'
else:
row['license_plate'] = plate
else:
row['license_plate'] = '无牌'
row['unit_id'] = 0
if msg['StorageUrl1'] and msg['StorageUrl1'] != None:
row['has_image'] = 1
row['image_url'] = msg['StorageUrl1']
else:
row['has_image'] = 0
row['image_url'] = ''
row['region_id'] = msg['TollgateID'][:6]
if 'PlateClass' in msg and msg['PlateClass'] and msg["PlateClass"] != '':
row["plate_type_id"]=int(msg["PlateClass"])
else:
row["plate_type_id"]=0
if msg['TollgateID'] is None:
return None
row['location_id'] = base64.b64encode(msg['TollgateID'])
row['loc_id'] = msg['TollgateID']
row['device_id'] = base64.b64encode(msg['DeviceID'])
row['dev_id'] = msg['DeviceID']
row['lane_id'] = int(msg['LaneNo'])
if msg.get('Speed',0):
row['speed'] = int(msg['Speed'])
else:
row['speed'] = 0
ct = list(msg['PassTime'])
ct.insert(12,':')
ct.insert(10,':')
ct.insert(8,' ')
ct.insert(6,'-')
ct.insert(4,'-')
row['capture_time'] = ''.join(ct)
logging.error(row['capture_time'])
if msg['Direction']:
row['direction_id'] = str(msg['Direction'])
else:
row['direction_id'] = '0'
if row['location_id'] in self.location_to_location:
row['location_id'] = self.location_to_location[row['location_id']]
else:
#pass
query = "SELECT Name,Latitude,Longitude FROM `databasess`.`Tollgates` WHERE `TollgateID` = '%s' LIMIT 1 " % row['loc_id']
logging.info(query)
rs = self.execute_sql(query)
logging.info(rs)
if rs != None or len(rs)>0:
if rs[0]:
#logging.info("INSERT INTO mon_location SET location_name = '%s',loc_id='%s',region_code='%s',LONGITUDE='%s',LATITUDE='%s';" % (rs[0]['Name'].encode('utf-8'),base64.b64decode(row['location_id']).encode("utf-8"),row['region_id'],rs[0]['Longitude'],rs[0]['Latitude']))
insert_id = self.execute_sql("INSERT INTO mon_location SET location_name = '%s',loc_id='%s',region_code='%s',LONGITUDE='%s',LATITUDE='%s';" % (rs[0]['Name'].encode('utf-8'),base64.b64decode(row['location_id']).encode("utf-8"),row['region_id'],rs[0]['Longitude'],rs[0]['Latitude']),"insert")
else:
insert_id = self.execute_sql("INSERT INTO mon_location SET location_name = '%s',loc_id='%s',region_code='%s';" % (''.encode('utf-8'),base64.b64decode(row['location_id']).encode("utf-8"),row['region_id']),"insert")
if insert_id > 0:
#print insert_id,base64.b64decode(row['location_id']).encode("utf-8")
r = requests.get('http://127.0.0.1:9002/?format=json&action=update_location_id')
r = requests.get('http://127.0.0.1:9002/?format=json&action=make_location_dict_cache')
self.load_cache()
if row['location_id'] in self.location_to_location:
row['location_id'] = self.location_to_location[row['location_id']]
if row['device_id'] in self.device_to_device:
row['device_id'] = self.device_to_device[row['device_id']]
else:
#pass
insert_id = self.execute_sql("INSERT INTO mon_device SET dev_id = '%s',loc_id='%s',region_code='%s';" % (base64.b64decode(row['device_id']).encode("utf-8"), row['loc_id'],row['region_id']),"insert")
if insert_id > 0:
# print insert_id,base64.b64decode(row['device_id']).encode("utf-8"), row['loc_id']
r = requests.get('http://127.0.0.1:9002/?format=json&action=update_device_id')
r = requests.get('http://127.0.0.1:9002/?format=json&action=make_device_dict_cache')
self.load_cache()
if row['device_id'] in self.device_to_device:
row['device_id'] = self.device_to_device[row['device_id']]
return row
except Exception, e:
logging.exception('格式化信息时错误: %s', str(e))
return None
def run(self):
config_file = open(os.path.dirname(os.path.abspath(__file__)) + '/config.yaml')
self.config = yaml.safe_load(config_file)
config_file.close()
name = 'get_msg_from_redis_hik'
logging.basicConfig(level=logging.INFO)
handler = RotatingFileHandler('/var/log/%s.log' % name, maxBytes=134217728, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logging.getLogger('').addHandler(handler)
#-------------------同步输出到控制台-------------------
# console = logging.StreamHandler()
# console.setLevel(logging.INFO)
# formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# console.setFormatter(formatter)
# logging.getLogger().addHandler(console)
#-------------------------------------------------------
logging.warning('启动 [%s]', name)
# self.connected = False
self.db = None
self.connected = None
r = redis.StrictRedis(host='127.0.0.1', password=self.config['redis_mq']['password'],port=6380)
pipe = r.pipeline()
self.err_count = 0#统计设备id错误数
self.err_dev_time = {}#统计设备时间错误数
self.load_cache()
cache = pylru.lrucache(600000)
#self.pgconnect()
while True:
#if not self.connected:
# self.pgconnect()
try:
msg = r.lpop('_jobs_hik')#key的名称需要修改
if msg == None:
time.sleep(0.05)
continue
start = timeit.default_timer()
if msg :
if len(msg) == 0:
logging.info("msg eq 0")
continue
msg = json.loads(msg)
msg = msg['SubscribeNotificationListObject']['SubscribeNotificationObject'][0]['MotorVehicleObjectList']['MotorVehicleObject']
#print msg
for row in msg:
#print row
#rr.incr(time.strftime("%Y%m%d",time.localtime(time.time()))+":zqy")
msg = self.format_msg(row)
if msg:
#print msg
#break
try:
logging.info(msg)
r.incr(time.strftime("%Y%m%d",time.localtime(time.time()))+":yz")
pipe.lpush(self.config['redis_mq']['queue_key_name'], json.dumps(msg))
if pipe.__len__()==1:
start = timeit.default_timer()
except UnicodeDecodeError,ude:
logging.error('编码json时错误: %s',msg['license_plate'])
if timeit.default_timer()-start>1 or pipe.__len__()>100:#1秒或100条入redis一次
pass
pipe.execute()
except Exception,e:
logging.error('error:_____'+str(e))
if __name__ == "__main__":
daemon = MyDaemon('/var/run/get_msg_from_redis_hik.pid')
#daemon.run()
#sys.exit(0)
if len(sys.argv) == 2:
if 'start' == sys.argv[1]:
daemon.start()
elif 'stop' == sys.argv[1]:
daemon.stop()
elif 'restart' == sys.argv[1]:
daemon.restart()
else:
daemon.run()
print "Unknown command"
sys.exit(2)
sys.exit(0)
else:
print "usage: %s start|stop|restart" % sys.argv[0]
sys.exit(2)
抽取kafka数据到redis_mq模块 脚本
#!/usr/bin/python
#-*- coding: utf-8 -*-
# 抽取kafka数据到redis_mq模块
# 作者：树下水月
import MySQLdb
import redis
import requests
import json
import yaml,logging
from logging.handlers import TimedRotatingFileHandler,RotatingFileHandler
import time,timeit,datetime,sys,os,threading
import Queue
from daemon import Daemon
from kafka import KafkaConsumer
from kafka.structs import TopicPartition
import pickle
import base64
reload(sys)
sys.setdefaultencoding('utf-8')
class MyDaemon(Daemon):
def execute_sql(self,sql,action='select'):
try:
if self.db is None:
self.db = MySQLdb.connect(host=self.config['mysql_web']['host'],port=self.config['mysql_web']['port'], user=self.config['mysql_web']['user'], passwd=self.config['mysql_web']['password'], db=self.config['mysql_web']['database'],charset="utf8")
try:
self.db.ping()
except MySQLdb.Error,e:
self.db = MySQLdb.connect(host=self.config['mysql_web']['host'],port=self.config['mysql_web']['port'], user=self.config['mysql_web']['user'], passwd=self.config['mysql_web']['password'], db=self.config['mysql_web']['database'],charset="utf8")
mysql_web = self.db.cursor(MySQLdb.cursors.DictCursor)
r = mysql_web.execute(sql)
if action=='select':
r = mysql_web.fetchall()
elif action=='update':
pass
elif action=='insert':
r = self.db.insert_id()
mysql_web.close()
return r
except Exception, e:
logging.exception('连接数据库时错误: %s', str(e))
r = None
if action=='select':
r = []
elif action=='update':
pass
elif action=='insert':
r = 0
return r
def format_msg(self,msg):
try:
row = {}
if len(msg['license_plate'])<7 or msg['license_plate']=='未识别' or msg['license_plate']=='无' or msg['license_plate']=='无车牌':
row['license_plate'] = '无牌'
else:
row['license_plate'] = msg['license_plate']#.decode("gbk").encode('utf-8')
row['plate_type_id'] = msg['plate_type_id1'] if 'plate_type_id1' in msg else 0
row['region_id'] = msg['region_id']
#卡口编号
row['location_id'] = msg['location_id']
row['loc_id'] = msg['location_id']
#设备编号
row['device_id'] = msg['device_id']
row['dev_id'] = msg['device_id']
row['lane_id'] = msg['lane_id']
row['speed'] = msg['speed'] if 'speed' in msg else 0
row['direction_id'] = msg['direction_id']
#row['capture_time'] = str(msg['capture_time']).strftime('%Y-%m-%d %H:%M:%S')
row['capture_time'] = msg['capture_time']
if 'image_url1' in msg:
image_url = str(msg['image_url1'])
elif 'image_url' in msg:
image_url = str(msg['image_url'])
else:
image_url = ''
row['image_url'] = image_url
return row
except Exception, e:
logging.exception('格式化信息时错误: %s', str(e))
return None
def run(self):
config_file = open(os.path.dirname(os.path.abspath(__file__)) + '/config.yaml')
self.config = yaml.safe_load(config_file)
config_file.close()
name = 'get_msg_kafka'
logging.basicConfig(level=logging.INFO)
handler = RotatingFileHandler('/var/log/%s.log' % name, maxBytes=134217728, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logging.getLogger('').addHandler(handler)
#-------------------同步输出到控制台-------------------
# console = logging.StreamHandler()
# console.setLevel(logging.INFO)
# formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# console.setFormatter(formatter)
# logging.getLogger().addHandler(console)
#-------------------------------------------------------
logging.warning('启动 [%s]', name)
concurrency_lock=threading.BoundedSemaphore(value=self.config['get_msg']['thread_num'])
self.db = None
self.err_count = 0#统计设备id错误数
self.err_dev_time = {}#统计设备时间错误数
self.crop_config = {
"420100999999999":{"crop_x":0,"crop_y":0,"crop_w":0,"crop_h":0},
}
self.MSG_QUEEN = Queue.Queue(0)
#接入数据的点位
try:
r = redis.StrictRedis(unix_socket_path=self.config['redis_mq']['unix_socket_path'], password=self.config['redis_mq']['password'])
pipe = r.pipeline()
logging.warning('创建连接Kafka...')
kafka_brokers = "kafka1:9092,kafka2:9092"
# 实例化消费者
consumer = KafkaConsumer('monitorMessage',bootstrap_servers=kafka_brokers, auto_offset_reset='latest', group_id='monitorMessage-2021060100001')
recv_number = 0
start = timeit.default_timer()
while 1:
try:
for msg in consumer:
recv_number += 1
# 消息内容
message = msg.value
offset = msg.offset # kafka偏移量
if recv_number%5000==0:
logging.warning('offset:%d,recv:%d',offset,recv_number)
#continue
row = json.loads(message)
#print(row)
#continue
msg = self.format_msg(row)
#logging.info(msg['location_id'])
if msg:
try:
logging.info(msg)
pipe.incr(time.strftime("%Y%m%d",time.localtime(time.time()))+":CAR")
pipe.rpush(self.config['redis_mq']['queue_key_name'], json.dumps(msg))#,ensure_ascii=False
if pipe.__len__()==1:
start = timeit.default_timer()
#self.MSG_QUEEN.put(msg)
except UnicodeDecodeError,ude:
logging.error('编码json时错误: %s',msg['license_plate'])
if timeit.default_timer()-start>1 or pipe.__len__()>100:#1秒或100条入redis一次
pipe.execute()
except KeyboardInterrupt:
logging.error('Ctrl+C,终止运行')
return
except Exception, e:
logging.exception('读取kafka时错误: %s', str(e))
time.sleep(10)
except Exception, e:
logging.exception('取数据时错误: %s', str(e))
sys.exit(0)
if __name__ == "__main__":
daemon = MyDaemon('/var/run/get_msg_kafka.pid')
#daemon.run()
#sys.exit(0)
if len(sys.argv) == 2:
if 'start' == sys.argv[1]:
daemon.start()
elif 'stop' == sys.argv[1]:
daemon.stop()
elif 'restart' == sys.argv[1]:
daemon.restart()
else:
print "Unknown command"
sys.exit(2)
sys.exit(0)
else:
print "usage: %s start|stop|restart" % sys.argv[0]
sys.exit(2)
抽取oracle数据到redis_mq模块
#!/usr/bin/python
#-*- coding: utf-8 -*-
# 抽取oracle数据到redis_mq模块
# 作者：王成
import redis
import requests
import json
import yaml,logging
from logging.handlers import TimedRotatingFileHandler,RotatingFileHandler
import time,timeit,datetime,sys,os,threading
from datetime import datetime,timedelta
import Queue
from daemon import Daemon
import traceback
import StringIO
import cx_Oracle
import pickle
import pylru
import pickle
import MySQLdb
import base64
reload(sys)
sys.setdefaultencoding('utf-8')
os.environ['NLS_LANG']="SIMPLIFIED CHINESE_CHINA.AL32UTF8"
class MyDaemon(Daemon):
def execute_sql(self,sql,action='select'):
try:
if self.db is None:
self.db = MySQLdb.connect(host=self.config['mysql_web']['host'],port=self.config['mysql_web']['port'], user=self.config['mysql_web']['user'], passwd=self.config['mysql_web']['password'], db=self.config['mysql_web']['database'],charset="utf8")
try:
self.db.ping()
except MySQLdb.Error,e:
self.db = MySQLdb.connect(host=self.config['mysql_web']['host'],port=self.config['mysql_web']['port'], user=self.config['mysql_web']['user'], passwd=self.config['mysql_web']['password'], db=self.config['mysql_web']['database'],charset="utf8")
mysql_web = self.db.cursor(MySQLdb.cursors.DictCursor)
r = mysql_web.execute(sql)
if action=='select':
r = mysql_web.fetchall()
elif action=='update':
pass
elif action=='insert':
r = self.db.insert_id()
mysql_web.close()
return r
except Exception, e:
logging.exception('连接数据库时错误: %s', str(e))
r = None
if action=='select':
r = []
elif action=='update':
pass
elif action=='insert':
r = 0
return r
def connect(self):
while not self.connected:
try:
self.connected = False
self.client = cx_Oracle.connect('get_sees/user@1.1.1.1:1521/orcl')
except cx_Oracle.DatabaseError, e:
#logging.error('Connecting error: %s', str(e))
print "Connecting error: "+str(e)
time.sleep(1)
else:
#logging.info('Connected to Oracle server %s', '10.53.11.34')
self.connected = True
break
def sqlSelect(self,sql,client):
try:
cr=client.cursor()
cr.execute(sql)
rs=cr.fetchall()
result = []
for row in rs:
tmp = {}
i = 0
for desc in cr.description:
tmp[desc[0]] = row[i]
i+=1
result.append(tmp)
cr.close()
return result
except cx_Oracle.DatabaseError, e:
#logging.error('sqlSelect error: %s', str(e))
print "sqlSelect error: "+str(e)
return None
def format_msg(self,msg):
try:
row = {}
if len(msg['HPHM'])<7 or msg['HPHM']=='未识别' or msg['HPHM']=='无' or msg['HPHM']=='无车牌' or msg['HPHM']=='车牌':
row['license_plate'] = '无牌'
else:
row['license_plate'] = msg['HPHM'].strip()
#临朐region
row['region_id'] = '370724'
row["plate_type_id"]=int(msg["PLATE_TYPE_ID"])
#卡口编号
row['loc_id'] = str(msg['KKID'])
#设备编号
row['dev_id'] = str(msg['KKID'])
row['lane_id'] = msg['LANE']
row['speed'] = int(msg['SPEED'])
row['capture_time'] = msg['PSSJ'].strftime('%Y-%m-%d %H:%M:%S')
#if int(msg['KKID']) in d['由东向西']:
#row['direction_id'] = '1'
#elif int(msg['KKID']) in d['由西向东']:
#row['direction_id'] = '2'
#elif int(msg['KKID']) in d['由南向北']:
#row['direction_id'] = '3'
#elif int(msg['KKID']) in d['由北向南']:
#row['direction_id'] = '4'
#else:
row['direction_id'] = '0'
row['image_url'] = msg['URL']
return row
except Exception, e:
logging.exception('格式化信息时错误: %s', str(e))
return None
def read_id(self,file_name):
all_the_text = '0'
try:
file_object = open(file_name,'r')
all_the_text = file_object.readline()
file_object.close()
except:
pass
return all_the_text
def write_id(self,file_name,id):
file_object = open(file_name, 'w')
try:
file_object.write(id)
finally:
file_object.close()
def run(self):
config_file = open(os.path.dirname(os.path.abspath(__file__)) + '/config.yaml')
self.config = yaml.safe_load(config_file)
config_file.close()
name = 'get_msg_from_oracle'
logging.basicConfig(level=logging.INFO)
handler = RotatingFileHandler('/var/log/%s.log' % name, maxBytes=134217728, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logging.getLogger('').addHandler(handler)
#-------------------同步输出到控制台-------------------
# console = logging.StreamHandler()
# console.setLevel(logging.INFO)
# formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# console.setFormatter(formatter)
# logging.getLogger().addHandler(console)
#-------------------------------------------------------
logging.warning('启动 [%s]', name)
self.connected = False
self.db = None
self.err_count = 0#统计设备id错误数
self.err_dev_time = {}#统计设备时间错误数
cache = pylru.lrucache(600000)
try:
self.db = None
r = redis.StrictRedis(unix_socket_path=self.config['redis_mq']['unix_socket_path'], password=self.config['redis_mq']['password'])
pipe = r.pipeline()
logging.warning('创建连接Redis...')
self.connect()
while 1:
if not self.connected:
self.connect()
try:
start_time = self.read_id('time_bh.log').strip()
stramptime = time.mktime(time.strptime(str(start_time),'%Y-%m-%d %H:%M:%S'))
query = "SELECT max(UPDATE_TIME) AS MAXTIME from HIKVISION.BMS_VEHICLE_PASS WHERE UPDATE_TIME > TO_DATE( '"+str(start_time)+"' , 'yyyy-mm-dd hh24:MI:ss')"
#print query
rs = self.sqlSelect(query,self.client)
if rs == None or len(rs)==0:
time.sleep(5)
continue
if rs[0]['MAXTIME'] is None:
time.sleep(5)
continue
max_time = rs[0]['MAXTIME']
t_max_time = time.mktime(max_time.timetuple())
if (t_max_time - stramptime) > 30:
max_time = datetime.strptime(start_time,'%Y-%m-%d %H:%M:%S') + timedelta(seconds=28)
print '有数据',max_time
else:
print '等待数据',max_time
time.sleep(5)
continue
min_time_str = str(start_time)[0:19]
max_time_str = str(max_time)[0:19]
self.write_id('time_bh.log',str(max_time_str))
query = "select t1.image_server_id SERVER,t1.VEHICLE_ID ID,t1.PLATE_TYPE PLATE_TYPE_ID,t3.CROSSING_ID KKID,t3.CROSSING_NAME KKMC,t1.LANE_ID LANE,t1.VEHICLE_SPEED SPEED,t1.PLATE_INFO HPHM,cast(t1.PASS_TIME as date) PSSJ,t1.PIC_VEHICLE URL from BMS_VEHICLE_PASS t1,BMS_CROSSING_INFO t3 where t1.CROSSING_ID=t3.CROSSING_ID AND T1.UPDATE_TIME > to_date('" + min_time_str + "','YYYY-MM-DD hh24:mi:ss') and T1.UPDATE_TIME <= to_date('" + max_time_str + "','YYYY-MM-DD hh24:mi:ss')"
#print query
rs = self.sqlSelect(query,self.client)
if rs == None or len(rs)==0:
print 'no data'
time.sleep(5)
continue
start = timeit.default_timer()
for row in rs:
start_time = time.time()
msg = self.format_msg(row)
if msg:
try:
#print msg
pipe.incr(time.strftime("%Y-%m-%d",time.localtime(time.time()))+":BH")
pipe.rpush(self.config['redis_mq']['queue_key_name'], json.dumps(msg))
if pipe.__len__()==1:
start = timeit.default_timer()
except UnicodeDecodeError,ude:
logging.error('编码json时错误: %s',msg['license_plate'])
if timeit.default_timer()-start>1 or pipe.__len__()>100:#1秒或100条入redis一次
pipe.execute()
except KeyboardInterrupt:
logging.error('Ctrl+C,终止运行')
return
except Exception, e:
logging.exception('处理数据时错误: %s', str(e))
time.sleep(10)
except Exception, e:
logging.exception('取数据时错误: %s', str(e))
sys.exit(0)
if __name__ == "__main__":
daemon = MyDaemon('/var/run/get_msg_from_oracle.pid')
daemon.run()
sys.exit(0)
#if len(sys.argv) == 2:
# if 'start' == sys.argv[1]:
# daemon.start()
# elif 'stop' == sys.argv[1]:
# daemon.stop()
# elif 'restart' == sys.argv[1]:
# daemon.restart()
# else:
# print "Unknown command"
# sys.exit(2)
# sys.exit(0)
#else:
# print "usage: %s start|stop|restart" % sys.argv[0]
# sys.exit(2)
web_tools 映射数据接口脚本
#!/usr/bin/python
#-*- coding: utf-8 -*-
# File Name: web_tools.py
# Description: 制作映射，字典，缓存文件
import MySQLdb
import json
import yaml,logging
from logging.handlers import TimedRotatingFileHandler,RotatingFileHandler
import time,timeit,datetime,sys,os
import traceback
import StringIO
import pickle
import web
import base64
from daemon import Daemon
render = web.template.render('templates/')
class Index:
def execute_sql(self,sql,action='select'):
'''数据库连接 & 执行SQL'''
try:
db = MySQLdb.connect(host=web.config['mysql_web']['host'],port=web.config['mysql_web']['port'], user=web.config['mysql_web']['user'], passwd=web.config['mysql_web']['password'], db=web.config['mysql_web']['database'],charset="utf8")
mysql_web = db.cursor(MySQLdb.cursors.DictCursor)
r = mysql_web.execute(sql)
if action=='select':
r = mysql_web.fetchall()
elif action=='update':
pass
elif action=='insert':
r = db.insert_id()
mysql_web.close()
db.close()
return r
except Exception, e:
logging.exception('连接数据库时错误: %s', str(e))
r = None
if action=='select':
r = []
elif action=='update':
pass
elif action=='insert':
r = 0
return r
def make_location_dict_cache(self):
'''制作 第三方卡口ID=>卡口ID 的映射字典'''
try:
all_location = self.execute_sql("SELECT location_id,loc_id FROM `mon_location` WHERE delete_flag = 0;",action='select')
d = {}
for location in all_location:
# print location['loc_id'],location['location_id']
d[base64.b64encode(location['loc_id'].encode("utf-8"))] = str(location['location_id'])
d_str = pickle.dumps(d)
f=file(os.path.dirname(os.path.abspath(__file__))+"/data/location_dict.dat","wb")
f.write(d_str)
f.close()
# with open('/opt/message/data/location_dict.dat') as f:
# d_str = f.read()
# print pickle.loads(d_str)
return True,len(d)
except Exception, e:
logging.exception('制作卡口缓存时错误: %s', str(e))
return False,0
def update_location_id(self):
'''根据数据库自增ID及区域编码更新卡口ID'''
try:
c = self.execute_sql("UPDATE mon_location as ml SET ml.location_id = ml.region_code*1000000+ml.id WHERE ml.location_id = 0 AND ml.region_code !=0;",action='update')
return True,c
except Exception, e:
logging.exception('更新卡口ID时错误: %s', str(e))
return False,0
def make_device_dict_cache(self):
'''制作 第三方设备ID=>设备ID 的映射字典'''
try:
all_device = self.execute_sql("SELECT device_id,dev_id FROM `mon_device` WHERE delete_flag = 0;",action='select')
d = {}
for device in all_device:
# print device['dev_id'],device['device_id']
d[base64.b64encode(device['dev_id'].encode("utf-8"))] = str(device['device_id'])
d_str = pickle.dumps(d)
f=file(os.path.dirname(os.path.abspath(__file__))+"/data/device_dict.dat","wb")
f.write(d_str)
f.close()
# with open('/opt/message/data/device_dict.dat') as f:
# d_str = f.read()
# print pickle.loads(d_str)
return True,len(d)
except Exception, e:
logging.exception('制作设备缓存时错误: %s', str(e))
return False,0
def update_device_id(self):
'''根据数据库自增ID及区域编码更新设备ID'''
try:
self.execute_sql("UPDATE mon_device as md,mon_location as ml SET md.location_id = ml.location_id WHERE md.location_id=0 AND md.loc_id = ml.loc_id;",action='update')
c = self.execute_sql("UPDATE mon_device as md SET md.device_id = md.location_id*1000000+md.id WHERE md.device_id = 0 AND md.location_id !=0;",action='update')
return True,c
except Exception, e:
logging.exception('更新设备ID时错误: %s', str(e))
return False,0
def make_device_to_location_dict_cache(self):
'''制作 设备ID=>卡口ID 的映射字典'''
try:
all_device = self.execute_sql("SELECT device_id,location_id FROM `mon_device` WHERE delete_flag = 0;",action='select')
d = {}
for device in all_device:
d[str(device['device_id'])] = str(device['location_id'])
d_str = pickle.dumps(d)
f=file(os.path.dirname(os.path.abspath(__file__))+"/data/device_to_location_dict.dat","wb")
f.write(d_str)
f.close()
return True,len(d)
except Exception, e:
logging.exception('制作设备到卡口缓存时错误: %s', str(e))
return False,0
def make_yearid_to_info_dict_cache(self):
'''制作 year_id 到 品牌型号年款 的映射字典'''
try:
all_info = self.execute_sql("SELECT cy.brandID,cy.modelID,cy.yearID,cm.levelID FROM `car_model` as cm,car_year as cy WHERE cm.modelID = cy.modelID;",action='select')
d = {}
for info in all_info:
d[info['yearID']] = {'brandID':info['brandID'],'modelID':info['modelID'],'levelID':info['levelID']}
d_str = pickle.dumps(d)
f=file(os.path.dirname(os.path.abspath(__file__))+"/data/yearid_to_info_dict.dat","wb")
f.write(d_str)
f.close()
return True,len(d)
except Exception, e:
logging.exception('制作year_id到品牌型号年款缓存时错误: %s', str(e))
return False,0
def make_car_lib_location_info_dict_cache(self):
'''制作 重点车辆库 停车场、收费站 的映射字典'''
try:
all_info = self.execute_sql("SELECT id,location_type FROM location WHERE location_type in (5,6);",action='select')
d = {}
for info in all_info:
d[info['id']] = {'location_type':info['location_type']}
d_str = pickle.dumps(d)
f=file(os.path.dirname(os.path.abspath(__file__))+"/data/car_lib_location_info_dict.dat","wb")
f.write(d_str)
f.close()
return True,len(d)
except Exception, e:
logging.exception('制作 重点车辆库 停车场、收费站 的映射字典: %s', str(e))
return False,0
def make_car_lib_area_info_dict_cache(self):
'''制作 重点车辆库 重点区域 的映射字典'''
try:
all_info = self.execute_sql("SELECT lib_id , keynote_location FROM car_lib_set ",action='select')
d = {}
for info in all_info:
d[info['lib_id']] = {'keynote_location':info['keynote_location']}
d_str = pickle.dumps(d)
f=file(os.path.dirname(os.path.abspath(__file__))+"/data/car_lib_area_info_dict.dat","wb")
f.write(d_str)
f.close()
return True,len(d)
except Exception, e:
logging.exception('制作 重点车辆库 重点区域 的映射字典: %s', str(e))
return False,0
def format_output(self,r,c,msg,format='html'):
'''格式化输出信息'''
if r:
if format == 'html':
return render.index(state=1, msg=msg)
else:
return json.dumps({'state':r,'count':c})
else:
if format == 'html':
return render.index(state=1, msg="失败")
else:
return json.dumps({'state':r,'count':c})
def GET(self):
web.header("Content-Type","text/html; charset=utf-8")
#print web.input()
user_data = web.input(action=None,format='html')
if user_data.action=='make_location_dict_cache':
r,c = self.make_location_dict_cache()
return self.format_output(r,c,msg="成功,字典长度%d,请重启get_msg脚本" % c,format=user_data.format)
elif user_data.action=='update_location_id':
r,c = self.update_location_id()
return self.format_output(r,c,msg="成功,更新%d行" % c,format=user_data.format)
elif user_data.action=='make_device_dict_cache':
r,c = self.make_device_dict_cache()
return self.format_output(r,c,msg="成功,字典长度%d,请重启get_msg脚本" % c,format=user_data.format)
elif user_data.action=='update_device_id':
r,c = self.update_device_id()
return self.format_output(r,c,msg="成功,更新%d行" % c,format=user_data.format)
elif user_data.action=='make_device_to_location_dict_cache':
r,c = self.make_device_to_location_dict_cache()
return self.format_output(r,c,msg="成功,字典长度%d" % c,format=user_data.format)
elif user_data.action=='make_yearid_to_info_dict_cache':
r,c = self.make_yearid_to_info_dict_cache()
return self.format_output(r,c,msg="成功,字典长度%d" % c,format=user_data.format)
elif user_data.action=='make_car_lib_location_info_dict_cache':
r,c = self.make_car_lib_location_info_dict_cache()
return self.format_output(r,c,msg="成功,字典长度%d" % c,format=user_data.format)
elif user_data.action=='make_car_lib_area_info_dict_cache':
r,c = self.make_car_lib_area_info_dict_cache()
return self.format_output(r,c,msg="成功,字典长度%d" % c,format=user_data.format)
else:
return render.index(state=0, msg="")
class MyDaemon(Daemon):
def run(self):
self.urls = (
'/', 'Index'
)
self.app = web.application(self.urls, globals())
try:
config_file = open(os.path.dirname(os.path.abspath(__file__)) + '/config.yaml')
web.config = yaml.safe_load(config_file)
config_file.close()
name = 'web_tools'
logging.basicConfig(level=logging.INFO)
handler = RotatingFileHandler('/var/log/%s.log' % name, maxBytes=134217728, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logging.getLogger('').addHandler(handler)
#-------------------同步输出到控制台-------------------
# console = logging.StreamHandler()
# console.setLevel(logging.INFO)
# formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# console.setFormatter(formatter)
# logging.getLogger('').addHandler(console)
#-------------------------------------------------------
logging.warning('启动 [%s]', name)
sys.argv[1] = web.config['web_tools']['listen']
logging.warning('监听 [%s]', sys.argv[1])
self.app.run()
except KeyboardInterrupt:
sys.exit()
except Exception, e:
logging.exception('启动 [%s] 错误: %s', name,str(e))
if __name__ == "__main__":
daemon = MyDaemon('/var/run/web_tools.pid')
#daemon.run()
#sys.exit()
if len(sys.argv) == 2:
if 'start' == sys.argv[1]:
daemon.start()
elif 'stop' == sys.argv[1]:
daemon.stop()
elif 'restart' == sys.argv[1]:
daemon.restart()
else:
daemon.run()
print "Unknown command"
sys.exit(2)
sys.exit(0)
else:
print "usage: %s start|stop|restart" % sys.argv[0]
sys.exit(2)
多进程多线程处理脚本
#!/usr/bin/python
#-*- coding: utf-8 -*-
#########################################################################
# File Name: worker_to_kafka.py
# Description: 数据处理脚本
# Version: 0.0.1
from datetime import datetime,date
import os,sys,time,timeit,random
import logging
from logging.handlers import TimedRotatingFileHandler,RotatingFileHandler
import multiprocessing,threading
import redis,Queue
import yaml,json,base64
import math
import requests
import contextlib
from urllib2 import urlopen
import pickle
import xxhash
import binascii
from confluent_kafka import Consumer
from confluent_kafka import Producer
from confluent_kafka import KafkaError
from daemon import Daemon
import random
import copy,shutil
reload(sys)
sys.setdefaultencoding('utf-8')
def redis_connect(redis_config):
if 'database' not in redis_config:
redis_config['database'] = 0
if 'password' not in redis_config:
redis_config['password'] = ''
if ('host' not in redis_config or str(redis_config['host']) == '127.0.0.1') and 'unix_socket_path' in redis_config:
redis_connect = redis.StrictRedis(unix_socket_path=redis_config['unix_socket_path'], password=redis_config['password'], db=redis_config['database'])
else:
redis_connect = redis.StrictRedis(host=redis_config['host'], port=redis_config['port'], password=redis_config['password'], db=redis_config['database'])
return redis_connect
def generate_info_id(capture_time, image_url):
"""消息唯一uuid生成
Args:
capture_time: 拍摄时间, 格式为unix时间戳
image_url: 图片url
Returns:
返回uuid字符串
"""
# capture_time_hex = binascii.b2a_hex(struct.pack('>I', clean_info['capture_time'])).decode()
capture_time_hex = hex(capture_time)[2:]
if image_url:
unique_hash = (xxhash.xxh64(image_url).hexdigest() +
binascii.b2a_hex(os.urandom(4)))
else:
unique_hash = binascii.b2a_hex(os.urandom(12))
return '%s-%s-%s-%s-%s' % (capture_time_hex,
unique_hash[:4],
unique_hash[4:8],
unique_hash[8:12],
unique_hash[12:])
class Dispatcher(multiprocessing.Process):
def __init__(self,config,cpu_count):
multiprocessing.Process.__init__(self)
self.daemon = True
self.config = config#配置文件
self.thread_num = int(math.ceil(float(config['worker']['thread_num'])/cpu_count))#计算出每个进程开启的线程数
self.threads = []#处理图像的线程数组
self.try_times = 5#重试次数
self.queen_size = 500#消息队列最大数量
self.message_queen = Queue.Queue(0)#消息队列
self.result_queen = Queue.Queue(0)#结果队列
self.car_face_queen = Queue.Queue(0)#结果队列
self.weed_server = {} #weed存储列表
self.weed_server_error_num = 10 #weed存储连续错误次数，超过该次数以后停用该weed，到下一小时再进行检测
self.post_video_processing_error_num = 100 #推送数据到video_processing连续错误次数，超过该次数以后将数据保存到文件，待后续再推
self.sreq = requests.Session()#利用会话减少网络连接开销
a = requests.adapters.HTTPAdapter(pool_connections = 300, pool_maxsize = 300,max_retries=1)
self.sreq.mount('http://', a)
self.yearid2info = {}#车辆年款<=>车辆品牌,型号,类型映射
year_file = os.path.dirname(os.path.abspath(__file__))+"/data/yearid_to_info_dict.dat"
if os.path.isfile(year_file):
with open(year_file) as f:
d_str = f.read()
self.yearid2info = pickle.loads(d_str)
else:
logging.error('加载yearID映射字典时错误: 未能找到指定文件')
def run(self):
logging.info('%s pid %d', self.name,os.getpid())
self.redis_num = redis_connect(self.config['redis_quick'])
#可用的weed集群，调用错误大于一定数量以后停止使用错误的服务器，一小时以后重置状态
self.weed_server = []
k = 0
for face_weed in self.config['worker']['face_weed']['host']:
t = {}
t['url'] = face_weed
t['error_num'] = 0 #错误次数，连续失败时才会记录
t['status'] = 1 #weed状态，1正常，0不可用
t['k'] = k
k += 1
self.weed_server.append(t)
# 每个进程开启1个读数据线程
GMT = threading.Thread(target=self.get_message)
self.threads.append(GMT)
# 每个进程开启1个保存数据线程
t = threading.Thread(target=self.save_result_to_kafka)
self.threads.append(t)
if self.config['worker']['save_carface_to_vehicle_srvdb'] and self.config['worker']['save_carface_image_to_weed']:
#定时检查weed集群状态
t = threading.Thread(target=self.check_weed_status)
self.threads.append(t)
logging.warning('CPU线程数是 [%d]', self.thread_num)
for x in xrange(0, self.thread_num):
t = threading.Thread(target=self.processing,name = self.name+'>processing-'+str(x))
self.threads.append(t)
for t in self.threads:
t.start()
time.sleep(1)
for t in self.threads:
t.join()
def get_message(self):#从redis抽取数据
redis_mq = redis_connect(self.config['redis_mq'])
pipe = redis_mq.pipeline()
while 1:
if self.queen_size > self.message_queen.qsize():
try:
for x in xrange(10):
pipe.lpop((self.config['redis_mq']['queue_key_name']))
messages = pipe.execute()
#logging.info(messages)
except Exception,e:
logging.exception('从redis抽取数据时错误: %s', str(e))
else:
for msg in messages:
if msg:
row = json.loads(msg)
logging.info(row)
# self.redis_num.incr(str(row['capture_time'].replace('-','').replace(' ','')[:10])+":from_mq")
self.message_queen.put({'message':msg,'try_times':0})
else:
time.sleep(0.05)
else:
#print 'message_queen is full'
time.sleep(1)
def save_result_to_kafka(self):
"""将数据保存到kafka"""
def delivery_callback(err, msg):
if err:
logging.error('Message failed delivery: %s' % err)
#else:
# logging.warning('Message delivered to {} [{}]'.format(msg.topic(),msg.partition()))
producer_conf = {
'bootstrap.servers': ','.join(self.config['kafka']['host'])
}
# start_time = timeit.default_timer()
while 1:
try:
producer = Producer(**producer_conf)
#print(producer)
while 1:
try:
result = self.result_queen.get(timeout=5)
try:
if 'time_int' in result:
del result['time_int']
producer.produce(
self.config['kafka']['topic_in'], json.dumps(result),
callback=delivery_callback)
#logging.info(result['source_id'])
#logging.info(result)
# self.redis_num.incr(str(result['capture_time'].replace('-','').replace(' ','')[:10])+":save_kafka")
# if timeit.default_timer() - start_time > 1:
logging.info('推送kafka成功: %s', result)
# start_time = timeit.default_timer()
except BufferError as e:
logging.error('kafka本地队列已满（%d 条），尝试重新写入kafka' % len(producer))
self.result_queen.put(result)
time.sleep(1)
except Exception as e:
logging.exception('保存kafka时错误: %s', str(e))
self.result_queen.put(result)
time.sleep(0.1)
producer.poll(0)
except Queue.Empty:
time.sleep(0.01)
continue
except Exception as e:
logging.exception('连接kafka时错误: %s', str(e))
time.sleep(10)
def processing(self):#分析过车数据
thread = threading.current_thread()
logging.warning('线程ID [%s]', thread.getName())
while 1:
try:
message_info = self.message_queen.get(timeout=10)
message = message_info['message']
try_times = message_info['try_times']+1
try:
json_msg = json.loads(message)
except Exception, e:
logging.exception('转json对象时错误: %s %s', str(e), message)
else:
image_url = json_msg['image_url']
img_data = ''
#start = timeit.default_timer()
try:
down_time1 = time.time()
if 0:
img_data = ''
elif image_url.startswith('http'):
#r = self.sreq.get(image_url, timeout=2)
if "127.0.0.1" in image_url:
image_url = ''
else:
r = requests.get(image_url, timeout=10)
img_data = r.content
elif image_url.startswith('ftp'):
with contextlib.closing(urlopen(image_url, None, 10)) as r:
img_data = r.read()
elif image_url.startswith('/data'):
file_obj=open(image_url)
img_data = file_obj.read()
file_obj.close()
else:
if len(image_url)>0:
logging.error('无效的图片地址: [%s]', image_url)
down_time2 = time.time()
down_loss_time = down_time2 - down_time1
if down_loss_time > 3:
logging.warning("下载图片耗时较长: {:.2f}, url: {}".format(down_loss_time, image_url))
except Exception, e:
#if try_times < self.try_times:
#self.message_queen.put({'message':message,'try_times':try_times})#将消息退回队列
#continue
logging.warning('下载图片时错误: [%s] %s', image_url,str(e))
# self.redis_num.incr(str(json_msg['capture_time'].replace('-','').replace(' ','')[:10])+":down_img")
#print 'download image',(timeit.default_timer() - start)
result = {}
result['capture_time'] = datetime.strptime(json_msg['capture_time'], '%Y-%m-%d %H:%M:%S')
result['capture_time'] = result['capture_time'].strftime('%Y-%m-%d %H:%M:%S')
result['time_int'] = int(time.mktime(time.strptime(json_msg['capture_time'], '%Y-%m-%d %H:%M:%S')))
result['region_id'] = int(json_msg['region_id'])
if 'city_name' in json_msg:
result['city_name'] = json_msg['city_name']
else:
result['city_name'] = 'other'
result['loc_id'] = str(json_msg['loc_id'])
if 'location_id' in json_msg:
result['location_id'] = str(json_msg['location_id'])
result['dev_id'] = str(json_msg['dev_id'])
if 'device_id' in json_msg:
result['device_id'] = str(json_msg['device_id'])
result['is_face'] = 0
result['source_id'] = 22 #默认source_id为22
result['image_url1'] = json_msg['image_url']
result['image_url'] = json_msg['image_url']
result['license_plate1'] = json_msg['license_plate']
result['license_plate2'] = '未识别'
result['plate_type_id1'] = int(json_msg['plate_type_id'])
result['speed'] = int(json_msg['speed'])
result['capture_type'] = 0
result['lane_id'] = int(json_msg['lane_id'])
if str(json_msg['direction_id']).isdigit():
result['direction_id'] = int(json_msg['direction_id'])
else:
result['direction_id'] = 0
result['info_id'] = generate_info_id(int(time.mktime(time.strptime(json_msg['capture_time'], '%Y-%m-%d %H:%M:%S'))),json_msg['image_url'])
img_len = len(img_data)
if img_len>10000 and img_len<100000000:
# self.redis_num.incr(str(json_msg['capture_time'].replace('-','').replace(' ','')[:10])+":to_fusion")
payload = { 's33witc1h_od':1,
'switc12h_vehprop':1,
}
json_data = []
try:
files = {'image_file': ('image.jpg', img_data)}
recogn_time1 = time.time()
#logging.info(payload)
r = self.sreq.post(self.config['fcgi']['get_api_url'], files=files, data=payload)
recogn_time2 = time.time()
recogn_loss_time = recogn_time2 - recogn_time1
if recogn_loss_time > 2:
logging.info("识别接口耗时较长: {:.2f}".format(recogn_loss_time))
if r.status_code == requests.codes.ok:
json_data = r.json()
#logging.info(json.dumps(json_data))
else:
logging.error('识别接口异常: [%s]', image_url)
except Exception, e:
logging.error('分析图片时错误: [%s] %s', image_url,str(e))
if try_times < self.try_times:
self.message_queen.put({'message':message,'try_times':try_times})#将消息退回队列
# self.redis_num.incr(str(json_msg['capture_time'].replace('-','').replace(' ','')[:10])+":from_fusion")
#识别结果有问题时将数据直接入库
if not json_data or 'data' not in json_data or 'objects' not in json_data['data'] or len(json_data['data']["objects"]) == 0:
while self.queen_size < self.result_queen.qsize():
time.sleep(2)
logging.warning('结果队列已满1: [%s]', thread.getName())
self.result_queen.put(result) # 放入保存数据队列
continue
else:
while self.queen_size < self.result_queen.qsize():
time.sleep(2)
logging.warning('结果队列已满3: [%s]', thread.getName())
self.result_queen.put(result)#放入保存数据队列
# self.redis_num.incr(str(json_msg['capture_time'].replace('-','').replace(' ','')[:10])+":no_img")
#logging.info(result)
except Queue.Empty:
#print 'message_queen is empty'
continue
except Exception,e:
logging.exception('分析过车数据时错误: %s', str(e))
#检查weed集群状态
def check_weed_status(self):
first_start = 1
while True:
try:
#重置可用的weed集群状态
for face_weed in self.weed_server:
#检查状态为0的weed集群，初次启动时检查所有的
if face_weed['status'] == 0 or first_start == 1:
weed_url = 'http://'+face_weed['url']+'/cluster/status?pretty=y'
try:
res = self.sreq.get(weed_url, timeout=5)
if res.status_code == requests.codes.ok:
self.weed_server[face_weed['k']]['status'] = 1
self.weed_server[face_weed['k']]['error_num'] = 0
else:
self.weed_server[face_weed['k']]['status'] = 0
self.weed_server[face_weed['k']]['error_num'] = 0
except Exception,e:
logging.error('weed集群%s状态异常', face_weed['url'])
self.weed_server[face_weed['k']]['status'] = 0
self.weed_server[face_weed['k']]['error_num'] = 0
logging.info('weed集群状态 %s', json.dumps(self.weed_server))
first_start = 0
time.sleep(60)
except Exception,e:
logging.error('检查weed集群状态错误: %s', str(e))
time.sleep(3)
class MyDaemon(Daemon):
def run(self):
config_file = open(os.path.dirname(os.path.abspath(__file__)) + '/config.yaml')
config = yaml.safe_load(config_file)
config_file.close()
name = 'worker_cpu'
logging.basicConfig(level=logging.INFO)
handler = RotatingFileHandler('/var/log/%s.log' % name, maxBytes=134217728, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(lineno)d- %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logging.getLogger().addHandler(handler)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.ERROR)
logging.warning('启动 [%s]', name)
logging.warning('主线程 Pid [%d]', os.getpid())
#处理入库失败文件后缀名,将带有.dat.r，.dat.tmp后缀的都改为.dat后缀
if config['worker']['post_carface_to_video_processing']:
try:
logging.info("处理入库失败文件后缀名")
failed_directory = config['worker']['failed_directory']
failed_files = os.listdir(failed_directory)
if failed_files:
for failed_file in failed_files:
filepath = os.path.join(failed_directory, failed_file)
#修改读取未完成的文件的后缀名
if failed_file.endswith('.dat.r'):
filepath_r = failed_file.replace(".dat.r", '.dat')
try:
shutil.move(filepath, filepath_r)
except Exception,e:
logging.exception('移动.dat.r失败文件时错误: %s', str(e))
#修改临时数据文件后缀名
if failed_file.endswith('.dat.tmp'):
if os.path.getsize(filepath) > 0:
filepath_r = failed_file.replace(".dat.tmp", '.dat')
try:
shutil.move(filepath, filepath_r)
except Exception,e:
logging.exception('移动.dat.tmp失败文件时错误: %s', str(e))
else:
try:
os.remove(filepath)
except Exception,e:
logging.exception('删除.dat.tmp失败文件时错误: %s', str(e))
time.sleep(2)
except Exception,e:
logging.exception('处理失败文件时错误: %s', str(e))
process = []
cpu_count = multiprocessing.cpu_count()
cpu_count = min(config['worker']['max_process'],cpu_count)
logging.warning('CPU线程数 [%d]', cpu_count)
for x in xrange(0, cpu_count):
process.append(Dispatcher(config,cpu_count))
for p in process:
p.start()
try:
for p in process:
p.join()
except KeyboardInterrupt:
for p in process:
p.terminate()
logging.warning('Ctrl+C,终止运行')
#=================本机测试代码用======================
if __name__ == "__main__":
daemon = MyDaemon('/var/run/worker_cpu.pid')
#daemon.run()
#sys.exit(1)
if len(sys.argv) == 2:
if 'start' == sys.argv[1]:
daemon.start()
elif 'stop' == sys.argv[1]:
daemon.stop()
elif 'restart' == sys.argv[1]:
daemon.restart()
else:
print "Unknown command"
sys.exit(2)
sys.exit(0)
else:
print "usage: %s start|stop|restart" % sys.argv[0]
sys.exit(2)
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献3条内容
所有评论(0)