context
继续昨天的内容，今天把twisted，pika整理一下
Let’s do it
twisted用来处理socket，pika用来和rabbitMQ做交互
twisted
twisted的官方文档介绍了很多用法，一开始我还以为twisted是个网络框架，在踩了一些坑，成功集成了pika之后，才发现自己错了。它更像一个runloop(loop)
在loop中注册事件，注册循环，来完成异步操作，循环操作等。和嵌入式的main()里面的那个while(1)很像，我就称他为:大循环吧
from twisted.internet import reactor
# 大循环就这么运行
reactor.run()
在大循环里面做点儿其他的事情：
from twisted.internet import reactor,task
# 用center实例来监听端口号为8123的TCP，center的具体实现后面再说～
reactor.listenTCP(8123, center)
# 在大循环中每0.01秒执行一次handle_message(),参数是queue_object
l = task.LoopingCall(handle_message, queue_object)
l.start(0.01)
reactor.run()
那么言归正传，先用twisted来搭建一个tcp服务器，先直接贴个代码：
socketEntity.py:
import datetime
from twisted.internet.protocol import connectionDone
from twisted.protocols.basic import LineReceiver
class SocketEntity(LineReceiver):
def __init__(self, device_list, addr):
#全局socket列表的 引用
self.global_socket_list_reference = device_list
#更新时间
self.updateTime = datetime.datetime.now()
#地址
self.address = addr.host
#收到了数据
def dataReceived(self, data):
print data
#连接建立
def connectionMade(self):
print "new connection from {}".format(self.address)
self.sending_connection_status(True)
if self.address in self.global_socket_list_reference.keys():
pass
else:
self.global_socket_list_reference[self.address] = self
#连接丢失
def connectionLost(self, reason=connectionDone):
print "connection {} lost".format(self.address)
self.sending_connection_status(False)
if self.address in self.global_socket_list_reference.keys():
del self.global_socket_list_reference[self.address]
#更新远程数据库状态
def sending_connection_status(self, online):
pass
socketServer.py:
import json
from twisted.internet.protocol import Factory
from socketEntity import SocketEntity
from twisted.internet import reactor, protocol
from twisted.internet import defer
from twisted.internet import task
class DeviceCenter(Factory):
def __init__(self):
这个就是全局socket列表
self.devices = {}
def send_cmd(self, addr, cmd):
if addr in self.devices:
self.devices[addr].sendLine(cmd)
else:
print "no connection"
# 这个是必须要的
def buildProtocol(self, addr):
return SocketEntity(self.devices, addr)
center = DeviceCenter()
try:
#增加监听端口的事件
reactor.listenTCP(8123, center)
reactor.run()
except KeyboardInterrupt:
print "socket stop manually"
except Exception as e:
print "socket error:", e.message
我觉得，光靠注释应该差不多可以知道这个在干嘛。其实这个和twisted官方的tcp文档基本一样，如果发现跑不起来(因为删了pika相关的东西后，没有再调试)可以去官方文档重新搞一份
说说我的理解～
前面说到了twisted是个事件驱动的框架，所以会注册很多的事件，比如注册一个监听事件，注册一个延迟事件，注册一个循环事件～
注册监听事件reactor.listenTCP(8123, center)，我觉得这个方法可能注册了很多小事件，这些小事件会调用DeviceCenter中的buildProtocol()，以及相应Protocol中的一些函数，比如dataReceived,connectionMade,connectionLost等
pika
这个框架就是rabbitMQ client的python版本，通过这个库，可以轻松的发送message到rabbitMQ，也可以轻松的监听rabbitMQ
官方文档就写了很简单的用法
官方文档里面对连接认证这一块描述的还是挺简单的，但是通过pycharm的自动补全，或者查看源代码，就能知道如何设置连接的各种参数以及认证参数
官方sender.py
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
官方receive.py
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
channel.basic_consume(callback,
queue='hello',
no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
很明显，这里pika很牛逼，有他的start_consuming()会阻塞整个主线程，会一直监听queue，直到消费者出现异常，或者用户手动break。这个对于后面在数据库服务器做，根据rabbitMQ的消息来修改数据库中相应数据这个功能非常友好。基本来说，靠这个例子就行了。
twisted+pika
因此，pika的start_consuming()是万万不能用到twisted中的
好在，pika的官方文档刚刚好给出了twisted的消费者example
我也贴一遍：
# -*- coding:utf-8 -*-
import pika
from pika import exceptions
from pika.adapters import twisted_connection
from twisted.internet import defer, reactor, protocol,task
@defer.inlineCallbacks
def run(connection):
channel = yield connection.channel()
exchange = yield channel.exchange_declare(exchange='topic_link', exchange_type='topic')
queue = yield channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)
yield channel.queue_bind(exchange='topic_link',queue='hello',routing_key='hello.world')
yield channel.basic_qos(prefetch_count=1)
queue_object, consumer_tag = yield channel.basic_consume(queue='hello',no_ack=False)
#在主循环中增加一个read的任务，周期为0.01秒
l = task.LoopingCall(read, queue_object)
l.start(0.01)
@defer.inlineCallbacks
def read(queue_object):
#消费一下
ch,method,properties,body = yield queue_object.get()
if body:
print(body)
yield ch.basic_ack(delivery_tag=method.delivery_tag)
parameters = pika.ConnectionParameters()
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
d = cc.connectTCP('hostname', 5672)
# 这个我还没有去了解是干嘛的
d.addCallback(lambda protocol: protocol.ready)
d.addCallback(run)
reactor.run()
虽然我最后写的和这个不太一样，但是踩了一下午的坑，才把它做对了～
下面贴一下我的server的代码。这个socket server的作用是，http server收到了http request之后，向rabbitMQ发送了一个消息，内容是向某个设备(address)发送一条命令(cmd),twisted tcp server收到了rabbitMQ中的消息后，作出send_cmd的操作
socketServer.py:
import json
from twisted.internet.protocol import Factory
from socketEntity import SocketEntity
from twisted.internet import reactor, protocol
from twisted.internet import defer
from twisted.internet import task
# 注意这里需要新建一个mqConfigure.py 并把相应的设置参数写在里面
from mqConfigure import *
import pika
from pika.adapters import twisted_connection
class DeviceCenter(Factory):
def __init__(self):
self.devices = {}
存一个连接的引用保险保险
self.connection = None
def send_cmd(self, addr, cmd):
if addr in self.devices:
self.devices[addr].sendLine(cmd)
else:
print "no connection"
def buildProtocol(self, addr):
return SocketEntity(self.devices, addr)
@defer.inlineCallbacks
def handle_message(self, data):
# 获取消息，并处理消息
ch, method, properties, body = yield data.get()
load_data = json.loads(body)
addr = load_data.get("address")
cmd = str(load_data.get("cmd"))
self.send_cmd(addr, cmd)
yield ch.basic_ack(delivery_tag=method.delivery_tag)
@defer.inlineCallbacks
def setup_mq_listener(self, conn):
#这个时候rabbitMQ的tcp已经连接成功了，所以做后面的事情都没问题
channel = yield conn.channel()
yield channel.queue_bind(exchange='exchange', queue=MQ_QUEUE, routing_key=MQ_QUEUE)
yield channel.basic_qos(prefetch_count=1)
queue_object, consumer_tag = yield channel.basic_consume(queue=MQ_QUEUE, no_ack=False)
#因为start_consuming()是阻塞的，所以就使用twisted里面常用的task手段来实现一个监听的操作(也挺像轮询的)
l = task.LoopingCall(self.handle_message, queue_object)
l.start(0.01)
center = DeviceCenter()
try:
reactor.listenTCP(8123, center)
# 连接rabbitMQ的参数
parameters = pika.ConnectionParameters(host=MQ_HOST, port=MQ_PORT, virtual_host=MQ_VHOST,
credentials=pika.PlainCredentials(MQ_USERNAME, MQ_PASSWORD))
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
# 我把它理解为异步操作的task对象，可以增加回调
d = cc.connectTCP(MQ_HOST, MQ_PORT)
#我依然不知道这个是干嘛的
d.addCallback(lambda protocol: protocol.ready)
# connectTCP成功之后的回调，设置监听MQ
d.addCallback(center.setup_mq_listener)
reactor.run()
except KeyboardInterrupt:
print "socket stop manually"
print "cleared data"
except Exception as e:
print "socket error:", e.message
print "cleared data"
那天花了比较多的时间浪费在理解twisted的运行机制上，所以让我感觉twisted非常难的样子。现在整理了一下，感觉其实并不难～
End
今天差不多就到这粒
明天或者后天，来写写数据库服务器如何处理rabbitMQ的内容，并修改Django的数据库；以及把整个系统做成分布式时需要考虑的一些细节吧～
