web程序需要给IPHONE手机推送消息，移动端同事写好了一个LIB，但是这个LIB使用的是阻塞IO，在APPLE服务器返回前程序是阻塞的，他们用了多线程来解决这个问题，但是webpy运行在apache里，无法进行线程管理，所以就迫切需要一个异步的机制来解决这个问题，希望做到需要发送消息时。调用一个函数，把数据扔进管道或者消息队列后就立即返回，也不管数据是否真的出去了，它相当于生产者，再有一个程序从管道或者队列中读取，进行实际的发送，相当于消费者
但是python似乎只封装了IPC（UNIX/LINUX平台进程通信规范）的匿名管道，命名管道的API，并无消息队列，而且管道似乎不易操作，因此想使用第三方的消息队列工具，那么我老大进行了技术选型，最终选择了amqp，这个消息队列组件使用erlang编写，启动了一个socket服务器，程序通过socket进行入队和出队的操作，不过这个组件应该是提供了大量的库隐藏了通信的部分，使代码看起来就像调用函数进行队列操作
client.py 测试客户端 向队列写入消息
#!/usr/bin/env python# -*- coding: utf-8 -*-#client.pyimport sysimport timeimport jsonfrom amqplib import client_0_8 as amqpconn = amqp.Connection( host="localhost:5672", userid="guest", password="guest", virtual_host="/", insist=False)chan = conn.channel()i = 0while 1: #msg = amqp.Message('Message %d' % i) #笔记更新 ''' data = { "noteId" : 1, "recvUserId" : 2, "title" : "test", "updateUserName":"你好", "remindCount":10, "projectName":"11", "token":"b258f5e3809017e371009f32eba7d72fcf51165406ce011951811b53db15b414", "isPushed":0, "messageType":"NoteUpdated", } ''' ''' data = { "projectId": 1, "recvUserId": 2, "reqUserName": "testUser", "reqUserStatus": "pending", "remindCount": 10, "projectName":"11", "token":"b258f5e3809017e371009f32eba7d72fcf51165406ce011951811b53db15b414", "messageType":"Apply", } ''' data = { "projectId": 1, "recvUserId": 2, "reqUserName": "testUser", "reqUserStatus": "", "remindCount": 10, "projectName": "11", "token":"b258f5e3809017e371009f32eba7d72fcf51165406ce011951811b53db15b414", "messageType":"Review", } s = json.dumps(data,ensure_ascii=False) print s msg = amqp.Message(s) msg.properties["delivery_mode"] = 2 chan.basic_publish(msg, exchange="sorting_room", routing_key="testkey") i += 1 time.sleep(1) breakchan.close()conn.close()
server.py 读取队列
#!/usr/bin/env python# -*- coding: utf-8 -*-from amqplib import client_0_8 as amqpimport process conn = amqp.Connection( host="localhost:5672", userid="guest", password="guest", virtual_host="/", insist=False)chan = conn.channel()chan.queue_declare( queue="po_box", durable=True, exclusive=False, auto_delete=False)chan.exchange_declare( exchange="sorting_room", type="direct", durable=True, auto_delete=False,)chan.queue_bind( queue="po_box", exchange="sorting_room", routing_key="testkey")def recv_callback(msg): #TODO #print msg.body process.starup(msg.body)chan.basic_consume( queue='po_box', no_ack=True, callback=recv_callback, consumer_tag="testtag")while True: chan.wait()#chan.basic_cancel("testtag")#chan.close()#conn.close()
process.py 业务逻辑模块
#!/usr/bin/env python# -*- coding: utf-8 -*-import jsonimport timeimport threadfrom APNSWrapper import *#---------------------------------------------------------------------------------------------'''从外部的数据转换成为标准数据'''def convert(args): print args data = json.loads(args) return data'''线程入口'''def threadMain(args): data = convert(args) msg = messagesFactory(data) msg.fillMessages(data) msg.push() #---------------------------------------------------------------------------------------------'''业务入口'''def startup(args): thread.start_new_thread(threadMain,(args,))#---------------------------------------------------------------------------------------------'''消息工厂'''def messagesFactory(data): if data["messageType"] == "NoteUpdated": return NoteUpdateMessages() if data["messageType"] == "Apply": return ApplyMessages() if data["messageType"] == "Review": return ReviewMessages() #---------------------------------------------------------------------------------------------'''消息基类'''class Messages: def __init__(self): self.Messages = [] def makeMessage(self,data): pass def fillMessages(self,data): message = self.makeMessage(data) self.Messages.append(message) def push(self): try: wrapper = APNSNotificationWrapper('/etc/ck.pem', True) for message in self.Messages: wrapper.append(message) wrapper.notify() print 'send success' except Exception, e: raise#---------------------------------------------------------------------------------------------'''笔记更新消息类'''class NoteUpdateMessages(Messages): def makeMessage(self,data): token = data["token"] deviceToken = token.decode('hex') message = APNSNotification() message.token(deviceToken) for key in ['updateUserName','title']: if type(data[key]) is unicode: data[key] = data[key].encode('utf8') alert = '共%d篇更新 %s编辑了“%s”' % (int(data["remindCount"]), data["updateUserName"], data["title"]) if data["isPushed"] == 0: message.alert(alert) message.badge(int(data["remindCount"])) message.sound('qing.caf') property1 = APNSProperty("NOTE_ID", int(data["noteId"]) ) message.appendProperty(property1) property2 = APNSProperty("USER_ID", int(data["recvUserId"]) ) message.appendProperty(property2) property3 = APNSProperty("REMIND_TIME", int(time.time()) ) message.appendProperty(property3) message.appendProperty(APNSProperty("REMIND_TYPE", 1 )) return message#---------------------------------------------------------------------------------------------'''申请加入项目消息类'''class ApplyMessages(Messages): def makeMessage(self,data): token = data["token"] deviceToken = token.decode('hex') message = APNSNotification() message.token(deviceToken) alertMsg = '' status = data["reqUserStatus"] if status == 'pending': alertMsg = '%s申请加入%s群组，请审批' % (data["reqUserName"].encode('utf-8'), data["projectName"].encode('utf-8'),) elif status == 'active': alertMsg = '%s已加入%s群组' % (data["reqUserName"].encode('utf-8'), data["projectName"].encode('utf-8'),) elif status == 'deny': alertMsg = '%s被拒绝加入%s群组' % (data["reqUserName"].encode('utf-8'), data["projectName"].encode('utf-8'),) elif status == 'removed': alertMsg = '%s已从%s群组移除' % (data["reqUserName"].encode('utf-8'), data["projectName"].encode('utf-8'),) message.alert(alertMsg) message.badge(int(data["remindCount"])) message.sound('qing.caf') message.appendProperty(APNSProperty("PROJECT_ID", int(data["projectId"]) )) message.appendProperty(APNSProperty("USER_ID", int(data["recvUserId"]) )) message.appendProperty(APNSProperty("REMIND_TIME", int(time.time()) )) message.appendProperty(APNSProperty("REMIND_TYPE", 3 )) message.appendProperty(APNSProperty("REMIND_MSG", alertMsg )) return message#---------------------------------------------------------------------------------------------'''审核结果消息类'''class ReviewMessages(Messages): def makeMessage(self,data): token = data["token"] deviceToken = token.decode('hex') message = APNSNotification() message.token(deviceToken) alertMsg = '' status = data["reqUserStatus"] if status == 'active': alertMsg = '您已加入%s群组' % (data["projectName"].encode('utf-8'),) elif status == 'deny': alertMsg = '您被拒绝加入%s群组' % (data["projectName"].encode('utf-8'),) elif status == 'removed': alertMsg = '您已从%s群组移除' % (data["projectName"].encode('utf-8'),) message.alert(alertMsg) message.badge(int(data["remindCount"])) message.sound('qing.caf') message.appendProperty(APNSProperty("PROJECT_ID", int(data["projectId"]) )) message.appendProperty(APNSProperty("USER_ID", int(data["recvUserId"]) )) message.appendProperty(APNSProperty("REMIND_TIME", int(time.time()) )) message.appendProperty(APNSProperty("REMIND_TYPE", 2 )) message.appendProperty(APNSProperty("REMIND_MSG", alertMsg )) return message