Authored by wuxiao

批量提交模式最后修改

... ... @@ -20,7 +20,7 @@ rabbitmq_vhost = '/'
influxdb_host = '127.0.0.1'
--influxdb_host = '54.222.219.223'
influxdb_port = 8086
influxdb_timeout = 2
--[[
mongodb服务器
... ... @@ -29,7 +29,7 @@ mongodb_host = '127.0.0.1'
mongodb_port = 27017
mongodb_user = 'root'
mongodb_pass = 'yoho@Uniformpass'
mongodb_timeout = 3
mongodb_timeout = 2
--[[
mysql服务器
... ... @@ -46,7 +46,7 @@ mysql_pass = 'yoho@Uniformpass'
--http_mq_port = 9500
--worker_num = 3
--daemonize = false
multi_insert_limit = 10
multi_insert_limit = 200
multi_insert_timeout = 3
... ...
... ... @@ -76,9 +76,6 @@ while True:
measurement = Config.info_channel+time.strftime('_%Y%m%d')
json_body = dict(data['data'],**data['tags'])
if json_body['message'] == 'sdhfsodifjoisdfsoi':
print json_body
middleware = 'mongodb'
... ...
... ... @@ -54,7 +54,7 @@ def rabbitmq_connect():
#连接influxdb
def influxdb_connect():
_influxdb = InfluxDBClient(Config.influxdb_host, Config.influxdb_port, timeout=1)
_influxdb = InfluxDBClient(Config.influxdb_host, Config.influxdb_port, timeout=Config.influxdb_timeout)
return _influxdb
#连接mongodb
... ... @@ -92,22 +92,21 @@ def connect_close():
#数据入库定时器
def data_pull_timer(insert_data):
print time.time()
try:
for middleware,dbs in insert_data.items():
for db,measurements in dbs.items():
for measurement,queue in measurements.items():
try:
data_pull(queue,'Timer')
data_pull(queue)
except:
data_pull(queue,'Timer')
data_pull(queue)
except Exception as err:
print err
finally:
Timer(Config.multi_insert_timeout,data_pull_timer,[insert_data]).start()
#数据入库
def data_pull(queue,type = 'Normal'):
def data_pull(queue):
if queue.is_empty():
return
json_bodys = queue.get_all()
... ...