Как я могу использовать send_json с pyzmq PUB SUB

Question

Мне нужно отправить словарь как сообщение от издателя подписчикам. С шаблоном REQ/REP send_json и recv_json работают хорошо, но я не могу найти заклинание, которое работает для PUB/SUB. Надеюсь, дело не в том, что PUB / SUB может работать только с send() и recv().

Вот список для эксперимента, который я собрал:

"""
Experiments with 0MQ PUB/SUB pattern
"""
import os
import sys
import time
import zmq
from multiprocessing import Process
from random import sample, choice

import signal
def handler(signum, frame):
    """ Handler for SIGTERM """
    # kill the processes we've launched
    try:
        for name, proc in _procd.iteritems():
            if proc and proc.is_alive():
                proc.terminate()
    finally:            
        os._exit(0)

signal.signal(signal.SIGTERM, handler)  
PORT = 5566
TOPICS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"

def publisher():
    """ Randomly update and publish topics """
    context = zmq.Context()
    sock = context.socket(zmq.PUB)
    sock.bind("tcp://*:{}".format(PORT))
    ## Init a dict of topic:value pairs
    alltopics = dict()
    for char in TOPICS:
        alltopics[char] = time.time()

    while True:
        topic = choice(TOPICS)
        alltopics[topic] = time.time()
        ## THIS IS SENDING
        sock.send_json((topic, alltopics))
        print "Sent topic {}".format(topic)
        time.sleep(1)


def client(number, topics):
    """ 
    Subscribe to list of topics and wait for messages. 
    """
    context = zmq.Context()
    sock = context.socket(zmq.SUB)
    sock.connect("tcp://localhost:{}".format(PORT))
    for topic in topics:
        sock.setsockopt(zmq.SUBSCRIBE, topic)

    print "subscribed to topics {}".format(topics)    

    while True:
        ## THIS NEVER RETURNS
        print sock.recv_json() 

        ## ALREADY TRIED THIS. DOES NOT WORK  
        #topic, msg = sock.recv_json()
        #print  "Client{}  {}:{}".format(number, topic, msg[topic])

        sys.stdout.flush()

if __name__ == '__main__':
    _procd = dict()
    ## Launch publisher
    name = 'publisher'
    _procd[name] = Process(target=publisher, name=name)
    _procd[name].start()

    ## Launch the subscribers
    for n in range(10):
        name = 'client{}'.format(n)
        _procd[name] = Process(target=client,
                               name=name,
                               args=(n, sample(TOPICS,3)))
        _procd[name].start()



    ## Sleep until killed
    while True:
        time.sleep(1)    

И вот выход до момента, когда я убиваю родительский процесс

$ python pubsub.py
Sent topic Y
subscribed to topics ['B', 'Q', 'F']
subscribed to topics ['N', 'E', 'O']
subscribed to topics ['Y', 'G', 'M']
subscribed to topics ['G', 'D', 'I']
subscribed to topics ['D', 'Y', 'W']
subscribed to topics ['A', 'N', 'W']
subscribed to topics ['F', 'K', 'V']
subscribed to topics ['A', 'Q', 'X']
subscribed to topics ['S', 'Y', 'V']
subscribed to topics ['E', 'S', 'D']
Sent topic I
Sent topic N
Sent topic J
Sent topic I
Sent topic A
Sent topic T
Sent topic A
Sent topic K
Sent topic V
Sent topic E

Подписки и отправка, кажется, в порядке, но клиенты никогда ничего не печатают. Трассировки для клиентских процессов показывают, что они зависают при вызове sock.recv_json(). Моя первая попытка закомментирована. Это также висит.

Answer 1

Я все еще хотел бы, чтобы это работало с send_json() и recv_json(), но, по предложению Джейсона, работает следующее:

def mogrify(topic, msg):
    """ json encode the message and prepend the topic """
    return topic + ' ' + json.dumps(msg)

def demogrify(topicmsg):
    """ Inverse of mogrify() """
    json0 = topicmsg.find('{')
    topic = topicmsg[0:json0].strip()
    msg = json.loads(topicmsg[json0:])
    return topic, msg 

с этим в издателе ()

    sock.send(mogrify(topic, alltopics))

и это в клиенте ()

   topic, msg = demogrify(sock.recv())

Вот полный список с последующим примером вывода:

#!/usr/bin/env python
# coding: utf8 
"""
Experiments with 0MQ PUB/SUB pattern.
Creates a publisher with 26 topics (A, B, ... Z) and
spawns clients that randomly subscribe to a subset
of the available topics. Console output shows 
who subscribed to what, when topic updates are sent
and when clients receive the messages.
Runs until killed.
Author: Michael Ellis
License: WTFPL
"""
import os
import sys
import time
import zmq
from multiprocessing import Process
from random import sample, choice
import json

PORT = 5566
TOPICS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" # split into ['A', 'B', ... ]
PUBSLEEP = 0.01 # Sleep time at bottom of publisher() loop.
NCLIENTS = 10  # Number of clients spawned.
NSUBS = 3  # Number of topics each client subscribes to.

assert NSUBS <= len(TOPICS)

def mogrify(topic, msg):
    """ json encode the message and prepend the topic """
    return topic + ' ' + json.dumps(msg)

def demogrify(topicmsg):
    """ Inverse of mogrify() """
    json0 = topicmsg.find('{')
    topic = topicmsg[0:json0].strip()
    msg = json.loads(topicmsg[json0:])
    return topic, msg

def publisher():
    """ Randomly update and publish topics """
    context = zmq.Context()
    sock = context.socket(zmq.PUB)
    sock.bind("tcp://*:{}".format(PORT))
    ## Init a dict of topic:value pairs
    alltopics = dict()
    for char in TOPICS:
        alltopics[char] = time.time()

    while True:
        try:
            topic = choice(TOPICS)
            alltopics[topic] = time.time()
            sock.send(mogrify(topic, alltopics))
            print "Sent topic {}".format(topic)
            time.sleep(PUBSLEEP)
        except KeyboardInterrupt:
            sys.exit()

def client(number, topics):
    """ 
    Subscribe to list of topics and wait for messages. 
    """
    context = zmq.Context()
    sock = context.socket(zmq.SUB)
    sock.connect("tcp://localhost:{}".format(PORT))
    for topic in topics:
        sock.setsockopt(zmq.SUBSCRIBE, topic)

    print "subscribed to topics {}".format(topics)    

    while True:
        try:
            topic, msg = demogrify(sock.recv())
            print  "Client{}  {}:{}".format(number, topic, msg[topic])
            sys.stdout.flush()
        except KeyboardInterrupt:
            sys.exit()

_procd = dict()
def run():
    """ Spawn publisher and clients. Loop until terminated. """
    ## Launch publisher
    name = 'publisher'
    _procd[name] = Process(target=publisher, name=name)
    _procd[name].start()

    ## Launch the subscribers
    for n in range(NCLIENTS):
        name = 'client{}'.format(n)
        _procd[name] = Process(target=client,
                               name=name,
                               args=(n, sample(TOPICS, NSUBS)))
        _procd[name].start()


    ## Sleep until killed
    while True:
        time.sleep(1)          

if __name__ == '__main__':
    import signal
    def handler(signum, frame):
        """ Handler for SIGTERM """
        # kill the processes we've launched
        try:
            for _, proc in _procd.iteritems():
                if proc and proc.is_alive():
                    proc.terminate()
        finally:            
            sys.exit()

    signal.signal(signal.SIGTERM, handler)

    run()

Пример вывода

$ pubsub.py 
Sent topic Q
subscribed to topics ['R', 'G', 'S']
subscribed to topics ['J', 'K', 'C']
subscribed to topics ['L', 'B', 'P']
subscribed to topics ['X', 'Z', 'A']
subscribed to topics ['K', 'O', 'R']
subscribed to topics ['J', 'Z', 'T']
subscribed to topics ['R', 'G', 'P']
subscribed to topics ['Y', 'A', 'O']
subscribed to topics ['U', 'S', 'C']
subscribed to topics ['B', 'P', 'L']
Sent topic U
Client8  U:1407506576.27
Sent topic E
Sent topic A
Client3  A:1407506576.29
Client7  A:1407506576.29
Sent topic A
Client3  A:1407506576.31
Client7  A:1407506576.31
Sent topic G
Client0  G:1407506576.32
Client6  G:1407506576.32
Sent topic E
Sent topic B
Client2  B:1407506576.34
Client9  B:1407506576.34
Sent topic R
Client0  R:1407506576.35
Client6  R:1407506576.35
Client4  R:1407506576.35
Sent topic U
Client8  U:1407506576.36
...

Answer 2

Я верю, что вы после того, как send_multipart(), скорее, чем send_json(), На языке ZeroMQ сообщение с несколькими кадрами является сообщением, состоящим из нескольких частей, и первый кадр определяет "тему подписчика" для pub/sub. Из примеров, которые я видел, он также угадает тему от начала строки, если вы отправите только один кадр.