PyZMQ请求应答模式-观察者模式

前面讲过zmq的device,用来充当客户端与服务端的中间件,以增加灵活性，让服务端也变成可插拔。然而device是zmq封装好的，怎样才能一窥内部的数据流呢？看下图

一看这图就明白了，MonitoredQueue在创建Queue同时，还提供第3个PUB socket来发布途经这个Queue的进出信息。

import time
import zmq
from zmq.devices.basedevice import ProcessDevice
from zmq.devices.monitoredqueuedevice import MonitoredQueue
from zmq.utils.strtypes import asbytes
from multiprocessing import Process
import random
  
frontend_port = 5559  
backend_port = 5560  
monitor_port = 5562  
number_of_workers = 2

def monitordevice():  
    in_prefix=asbytes('in')  
    out_prefix=asbytes('out')  
    monitoringdevice = MonitoredQueue(zmq.XREP, zmq.XREQ, zmq.PUB, in_prefix, out_prefix)  
      
    monitoringdevice.bind_in("tcp://127.0.0.1:%d" % frontend_port)  
    monitoringdevice.bind_out("tcp://127.0.0.1:%d" % backend_port)  
    monitoringdevice.bind_mon("tcp://127.0.0.1:%d" % monitor_port)  
      
    monitoringdevice.setsockopt_in(zmq.RCVHWM, 1)  #对进入socket的消息设置高水位
    monitoringdevice.setsockopt_out(zmq.SNDHWM, 1) #对向外发送的消息设置高水位（最大缓存量）
    monitoringdevice.start()    
    print("Program: Monitoring device has started")

def server(backend_port):  
    print("Program: Server connecting to device") 
    context = zmq.Context()  
    socket = context.socket(zmq.REP)  
    socket.connect("tcp://127.0.0.1:%s" % backend_port)  
    server_id = random.randrange(1,10005)  
    while True:  
        message = socket.recv()  
        print("Server: Received - %s" % message)  
        socket.send_string("Response from server #%s" % server_id)
def client(frontend_port, client_id):  
    print("Program: Worker #%s connecting to device" % client_id)
    context = zmq.Context()  
    socket = context.socket(zmq.REQ)  
    socket.connect("tcp://127.0.0.1:%s" % frontend_port)  
    request_num = 1  
    socket.send_string("Request #%s from client#%s" % (request_num, client_id))  
    #  Get the reply.  
    message = socket.recv_multipart()  
    print("Client: Received - %s" % message)

def monitor():  
    print("Starting monitoring process") 
    context = zmq.Context()  
    socket = context.socket(zmq.SUB)  
    print( "Collecting updates from server...")  
    socket.connect("tcp://127.0.0.1:%s" % monitor_port)  
    socket.setsockopt(zmq.SUBSCRIBE, b"")  
    while True:  
        string = socket.recv_multipart()  
        print("Monitoring Client: %s" % string)
if __name__ == '__main__':
    monitoring_p = Process(target=monitordevice)  
    monitoring_p.start()    
    server_p = Process(target=server, args=(backend_port,))  
    server_p.start()    
    monitorclient_p = Process(target=monitor)  
    monitorclient_p.start()    
    time.sleep(2)     
  
    for client_id in range(number_of_workers):  
        Process(target=client, args=(frontend_port, client_id,)).start()

    time.sleep(10)
    server_p.terminate()
    monitorclient_p.terminate()
    monitoring_p.terminate()