python使用amqp的例子

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.py

import sys
import time

import json
from amqplib import client_0_8 as amqp

conn = amqp.Connection(
host="localhost:5672",
userid="guest",
password="guest",
virtual_host="/",
insist=False)
chan = conn.channel()

i = 0
while 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)
break

chan.close()
conn.close()

server.py 读取队列

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from amqplib import client_0_8 as amqp
import 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 json
import time
import thread
from 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