本段代码主要作用是httpsqs队列的消费端守护进程，从httpsqs中取出数据，放入mongodb
#!/usr/bin/pythonimportsys,time,json,loggingimportQueue, threading, datetimefromlib.base.daemonimportDaemonfromlib.queue.httpsqs.HttpsqsClientimportHttpsqsClientfromlib.db.DbMongodbimportDbMongodb
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
datefmt='%a, %d %b %Y %H:%M:%S',
filename='myapp.log',
filemode='w')
queue=Queue.Queue()
httpsqs= HttpsqsClient('192.168.0.218','1218','httpsqs.com')
db= DbMongodb('192.168.0.119','testdb')classThreadGetHttpSqs(threading.Thread):def__init__(self):
threading.Thread.__init__(self)
self.httpsqs=httpsqs
self.queue=queuedefrun(self):whileTrue:
data= self.httpsqs.get('logtest')ifdataisnotNone:
self.queue.put(data)
logging.info('get:id %s , tablename %s'%(self.getName(),data))else:
time.sleep(3)classThreadInsertDB(threading.Thread):def__init__(self):
threading.Thread.__init__(self)
self.queue=queue
self.db=dbdefrun(self):whileTrue:
chunk=self.queue.get()
s=json.loads(chunk)
tablename= s['table']
data= s['data']
self.db.save(tablename,data)
logging.info('insert:id %s , tablename %s'%(self.getName(),tablename))
self.queue.task_done()classMyDaemon(Daemon):def_run(self):whileTrue:foriinrange(2):
t=ThreadGetHttpSqs()#t.setDaemon(True)t.start()foriinrange(2):
b=ThreadInsertDB()#t.setDaemon(True)b.start()
#线程已经为永真循环，进程不能再循环
time.wait()if__name__=="__main__":
daemon= MyDaemon('/tmp/daemon-example.pid')iflen(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)