Python消息队列与HTTP客户端技术解析

### Python 消息队列与 HTTP 客户端技术解析 #### 1. Python 消息队列的运用 在开发应用程序时，为了协调不同部分的工作，消息队列是一种常用的机制。像 Web 服务器前端的线程，有时会调用数据库客户端或文件服务器来完成更繁重的工作，而请求 - 响应模式就很适合 RPC 技术。 #### 1.1 常见消息队列的实现 独立服务器常被用于实现流行的消息队列，像 AMQP 协议就是广泛使用的与语言无关的消息队列协议，有多种开源服务器支持，如 RabbitMQ、Apache Qpid 服务器等。不过，很多程序员不会直接学习消息协议，而是借助第三方库，例如使用 Django 框架的 Python 程序员，会使用 Celery 分布式任务队列，它还支持不同的后端服务，甚至可以用简单的 Redis 键值存储作为“消息队列”。 #### 1.2 ZeroMQ 消息队列示例 为了方便理解，这里介绍 ZeroMQ（MQ），它将消息处理智能分散到每个消息客户端程序中，无需集中式代理。以下是一个用 MQ 实现的通过蒙特卡罗方法估算圆周率的例子： ```python #!/usr/bin/env python3 # Programming in Python: The Basics # Small application that uses several different message queues import random, threading, time, zmq B = 32 # number of bits of precision in each random integer def ones_and_zeros(digits): """Express `n` in at least `d` binary digits, with no special prefix.""" return bin(random.getrandbits(digits)).lstrip('0b').zfill(digits) def bitsource(zcontext, url): """Produce random points in the unit square.""" zsock = zcontext.socket(zmq.PUB) zsock.bind(url) while True: zsock.send_string(ones_and_zeros(B * 2)) time.sleep(0.01) def always_yes(zcontext, in_url, out_url): """Coordinates in the lower-left quadrant are inside the unit circle.""" isock = zcontext.socket(zmq.SUB) isock.connect(in_url) isock.setsockopt(zmq.SUBSCRIBE, b'00') osock = zcontext.socket(zmq.PUSH) osock.connect(out_url) while True: isock.recv_string() osock.send_string('Y') def judge(zcontext, in_url, pythagoras_url, out_url): """Determine whether each input coordinate is inside the unit circle.""" isock = zcontext.socket(zmq.SUB) isock.connect(in_url) for prefix in b'01', b'10', b'11': isock.setsockopt(zmq.SUBSCRIBE, prefix) psock = zcontext.socket(zmq.REQ) psock.connect(pythagoras_url) osock = zcontext.socket(zmq.PUSH) osock.connect(out_url) unit = 2 ** (B * 2) while True: bits = isock.recv_string() n, m = int(bits[::2], 2), int(bits[1::2], 2) psock.send_json((n, m)) sumsquares = psock.recv_json() osock.send_string('Y' if sumsquares < unit else 'N') def pythagoras(zcontext, url): """Return the sum-of-squares of number sequences.""" zsock = zcontext.socket(zmq.REP) zsock.bind(url) while True: numbers = zsock.recv_json() zsock.send_json(sum(n * n for n in numbers)) def tally(zcontext, url): """Tally how many points fall within the unit circle, and print pi.""" zsock = zcontext.socket(zmq.PULL) zsock.bind(url) p = q = 0 while True: decision = zsock.recv_string() q += 1 if decision == 'Y': p += 4 print(decision, p / q) def start_thread(function, *args): thread = threading.Thread(target=function, args=args) thread.daemon = True # so you can easily Ctrl-C the whole program thread.start() def main(zcontext): pubsub = 'tcp://127.0.0.1:6700' ```