pyzmq學習筆記

def device_func():
    thread_num = 3
    context = zmq.Context()
    url_router = 'inproc://ping-server'
    url_dealer = 'tcp://*:5559'

    # Socket do cliente
    frontend = context.socket(zmq.ROUTER)  # 或者是 zmq.ROUTER zmq.XREP
    frontend.bind(url_dealer)

    # Socket do servidor
    backend = context.socket(zmq.DEALER)  # 或者是 zmq.DEALER zmq.XREQ
    backend.bind(url_router)

    for i in range(1, thread_num + 1):
        thread = threading.Thread(target=server_func, args=(i, url_router, context))
        thread.start()

    # property找不到 Device()報錯
    zmq.device(zmq.QUEUE, frontend, backend)

    frontend.close()
    backend.close()
    context.term()


def server_func(name, url_router, context):
    print(">>> start %d %s" % (name, '.' * 50))
    socket = context.socket(zmq.REP)
    socket.connect(url_router)
    while True:
        try:
            message = socket.recv()
            r_data = msgpack.unpackb(message, encoding='utf-8')
            print('server %d received:' % name, r_data)

            data = 'server %d send: %s' % (name, utils.time_now())
            print(data)
            s_data = msgpack.packb(data)
            socket.send(s_data)

        except:
            traceback.print_exc()
            socket.close()
            break

client端，鏈接device的frontend端（dealer）:

def client2():
    url_dealer = 'tcp://localhost:5559'
    context = zmq.Context()
    socket = context.socket(zmq.REQ)

    print("Collecting data from server…")
    socket.connect(url_dealer)
    pid = os.getpid()
    i = 0
    while 1:
        i += 1
        # data = input('\n>>:').strip()
        data = ('send - msg%d' % i)
        # print(data)
        s_data = msgpack.packb(data.encode())
        socket.send(s_data)

        ret = socket.recv()
        r_data = msgpack.unpackb(ret, encoding='utf-8')
        print('>>>received msg%d server info: %s \n' % (i, r_data))

以上一塊兒理解了zmq的device、router、dealer。

官網指出：function zmq.device(device_type, frontend, backend) Deprecated since version libzmq-3.2: Use zmq.proxy 取而代之的 zmq.proxy(frontend, backend, capture)

同時又有 class zmq.devices.Device(device_type=3, in_type=None, out_type=None) 測試了跑不起來

備註：

inproc 是zmq本地進程/線程的傳輸方式。能夠自定義。

zmq的Polling

主要功能應該是一個程序種須要創建多個socket鏈接時，須要接收數據，又要轉發數據等。參考如下連接：

官方文檔

http://pjwqq.iteye.com/blog/2260791

官方API翻譯的

我這裏主要是使用到client的超時重試功能。重現客戶端不斷超時重試發送數據，服務端不斷接收回復。數據大量冗餘重複。

server：同上，不過加了一句 sleep：

time.sleep(6) message = socket.recv()

client：

def conn_socket(url_dealer):
    """ 創建socket鏈接"""
    try:

        context = zmq.Context()
        socket = context.socket(zmq.REQ)
        socket.connect(url_dealer)

        poller = zmq.Poller()
        poller.register(socket, zmq.POLLIN)

        print('[%d] socket conn success' % os.getpid())
        return socket, poller

    except:
        traceback.print_exc()


def close_socket(socket, poller):
    """ 關閉socket鏈接"""
    try:
        socket.close()
        poller.unregister(socket)
    except:
        traceback.print_exc()


def client1():
    """ """
    url_dealer = 'tcp://localhost:5559'
    socket, poller = conn_socket(url_dealer)
    try:

        i = 0
        while 1:
            i += 1
            # data = input('\n>>:').strip()
            data = 'msg%d' % i
            print('send:%s' % data)
            # print(data)
            s_data = msgpack.packb(data.encode())
            socket.send(s_data)

            while True:
                # 超時後從新鏈接,參數是毫秒
                if poller.poll(5 * 1000):
                    ret = socket.recv()
                    r_data = msgpack.unpackb(ret, encoding='utf-8')
                    print('>>>received:%s' % r_data)
                    break
                else:
                    close_socket(socket, poller)
                    socket, poller = conn_socket(url_dealer)

                    print('*resend:%s' % data)
                    socket.send(s_data)

    except Exception as e:
        traceback.print_exc()
    finally:
        close_socket(socket, poller)
        pass