test_zmq_pub.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#    Copyright 2011 timger
#    +Author timger
#    +Gtalk&Email yishenggudou@gmail.com
#    +Msn yishenggudou@msn.cn
#    +Weibo @timger http://t.sina.com/zhanghaibo
#    +twitter @yishenggudou http://twitter.com/yishenggudou
#    Licensed under the MIT License, Version 2.0 (the "License");

import zmq
import time
import sys

def  test_pub():
    context = zmq.Context()
    socket = context.socket(zmq.PUSH)
    socket.hwm = 1000
    socket.connect('tcp://127.0.0.1:50001')
    #socket.bind('tcp://0.0.0.0:5555')
    tic = time.time()
    i = 0
    for line in sys.stdin:
        socket.send(line)
        i += 1
        if i % 10000 == 0:
            toc = time.time()
            print "sent 10k msgs in %.2fs" % (toc-tic)
            tic = time.time()

test_pub()

test_zmq_router.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#    Copyright 2011 timger
#    +Author timger
#    +Gtalk&Email yishenggudou@gmail.com
#    +Msn yishenggudou@msn.cn
#    +Weibo @timger http://t.sina.com/zhanghaibo
#    +twitter @yishenggudou http://twitter.com/yishenggudou
#    Licensed under the MIT License, Version 2.0 (the "License");

import zmq

def main():

    try:
        context = zmq.Context(1)
        # Socket facing clients
        frontend = context.socket(zmq.PULL)
        frontend.hwm = 1000
        frontend.bind("tcp://*:50001")
        
        # Socket facing services
        backend = context.socket(zmq.PUSH)
        backend.hwm = 1000
        backend.bind("tcp://*:50002")

        zmq.device(zmq.FORWARDER, frontend, backend)
    except Exception, e:
        print e
        print "bringing down zmq device"
    finally:
        pass
        frontend.close()
        backend.close()
        context.term()

if __name__ == "__main__":
    main()

test_zmq_sub.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#    Copyright 2011 timger
#    +Author timger
#    +Gtalk&Email yishenggudou@gmail.com
#    +Msn yishenggudou@msn.cn
#    +Weibo @timger http://t.sina.com/zhanghaibo
#    +twitter @yishenggudou http://twitter.com/yishenggudou
#    Licensed under the MIT License, Version 2.0 (the "License");

import zmq
from time import sleep

def  test_sub():
    context = zmq.Context()
    socket = context.socket(zmq.PULL)
    socket.connect('tcp://127.0.0.1:50002')
    #socket.connect('tcp://127.0.0.1:5555')
    i = 0
    while 1:
        n = len(socket.recv())
        if i % 10000 == 0:
            print i, n
        i += 1

test_sub()