python使用zeromq实现多进程简单poll复用

参考：https://learning-0mq-with-pyzmq.readthedocs.io/en/latest/pyzmq/multisocket/zmqpoller.html
实现zmqpolling.py.py

点击(此处)折叠或打开

1. import zmq
   
2. import time
   
3. import sys
   
4. import random
   
5. from multiprocessing import Process
   
6. 
   
7. def server_push(port="5556"):
   
8.     context = zmq.Context()
   
9.     socket = context.socket(zmq.PUSH)
   
10.     socket.bind("tcp://*:%s" % port)
    
11.     print("Running server on port: ", port)
    
12.     # serves only 5 request and dies
    
13.     for reqnum in range(10):
    
14.         if reqnum < 6:
    
15.             socket.send("Continue".encode("ascii") )
    
16.         else:
    
17.             socket.send("Exit".encode("ascii") )
    
18.             break
    
19.         time.sleep (1)
    
20. 
    
21. def server_pub(port="5558"):
    
22.     context = zmq.Context()
    
23.     socket = context.socket(zmq.PUB)
    
24.     socket.bind("tcp://*:%s" % port)
    
25.     publisher_id = random.randrange(0,9999)
    
26.     print ("Running server on port: ", port)
    
27.     # serves only 5 request and dies
    
28.     for reqnum in range(10):
    
29.         # Wait for next request from client
    
30.         topic = random.randrange(8,10)
    
31.         messagedata = "server#%s" % publisher_id
    
32.         print("%s %s" % (topic, messagedata))
    
33.         socket.send( ("%d %s" % (topic, messagedata)).encode("ascii") )
    
34.         time.sleep(1)
    
35. 
    
36. def client(port_push, port_sub):
    
37.     context = zmq.Context()
    
38.     socket_pull = context.socket(zmq.PULL)
    
39.     socket_pull.connect ("tcp://localhost:%s" % port_push)
    
40.     print ("Connected to server with port %s" % port_push)
    
41.     socket_sub = context.socket(zmq.SUB)
    
42.     socket_sub.connect ("tcp://localhost:%s" % port_sub)
    
43.     socket_sub.setsockopt(zmq.SUBSCRIBE, "9".encode("ascii") )
    
44.     print ("Connected to publisher with port %s" % port_sub)
    
45.     # Initialize poll set
    
46.     poller = zmq.Poller()
    
47.     poller.register(socket_pull, zmq.POLLIN)
    
48.     poller.register(socket_sub, zmq.POLLIN)
    
49.     # Work on requests from both server and publisher
    
50.     should_continue = True
    
51.     while should_continue:
    
52.         socks = dict(poller.poll())
    
53.         if socket_pull in socks and socks[socket_pull] == zmq.POLLIN:
    
54.             message = socket_pull.recv()
    
55.             print ("Recieved control command: %s" % message)
    
56.         if message == "Exit":
    
57.             print ("Recieved exit command, client will stop recieving messages")
    
58.             should_continue = False
    
59. 
    
60.         if socket_sub in socks and socks[socket_sub] == zmq.POLLIN:
    
61.             string = socket_sub.recv()
    
62.             topic, messagedata = string.split()
    
63.             print("Processing ... ", topic, messagedata)
    
64. 
    
65. 
    
66. 
    
67. if __name__ == "__main__":
    
68.     # Now we can run a few servers
    
69.     server_push_port = "5556"
    
70.     server_pub_port = "5558"
    
71.     Process(target=server_push, args=(server_push_port,)).start()
    
72.     Process(target=server_pub, args=(server_pub_port,)).start()
    
73.     Process(target=client, args=(server_push_port,server_pub_port,)).start()

运行：
#python3 zmqpolling.py.py