pynng 超快速上手，但不保熟

nng 是 zeromq 的后辈，nanomsg 的 下一代的消息处理库。
主要就是快，上手超快，消息收发也超快。
本机 python 3.9 + pynng 0.7.1 本地环回测试 10000次1G数据，只需要1秒就可以完成。

nng相比zeromq的目前发现优势，链接状态全自动维护，自动重连，掉线重连能快速恢复，数据发送不完整时能自动无限重发（默认情况），报头能全定义从而能支持http协议，速度飞快。

以下4个简单完整例子
_try_push_pull.py
_try_pair_pair.py
_try_pub_sub.py
_try_rep_req.py

pynng安装

pip install -U pynng

_try_push_pull.py

from pynng import nng from multiprocessing import Process import numpy as np import pickledef proc_send_data():'''push端只管发数据:return:'''sock = nng.Push0(dial='tcp://127.0.0.1:8855')for i in range(10000):data = np.random.uniform([128, 4, 512, 512])data = pickle.dumps(data)sock.send(data)sock.close()def proc_recv_data():'''pull端只管收数据:return:'''sock = nng.Pull0(listen='tcp://127.0.0.1:8855')for i in range(10000):r = sock.recv(block=True)data = pickle.loads(r)print(i, np.mean(data))sock.close()if __name__ == '__main__':p1 = Process(target=proc_recv_data, daemon=True)p2 = Process(target=proc_send_data, daemon=True)print('start')p1.start()p2.start()p1.join()p2.join()print('success')

_try_pair_pair.py

from pynng import nng from pynng.exceptions import TryAgain, Timeout from multiprocessing import Process import numpy as np import pickle import randomdef proc_send_data():'''pair端想发就发，想收就收如果发消息前，对端已经关掉了，那么将会卡在send函数那里，直到对端重新打开，或send_timeout超时:return:'''sock = nng.Pair0(dial='tcp://127.0.0.1:8855', recv_timeout=500, send_timeout=500)for i in range(10000):# 发送1G的数据。128*4*512*512*8/1024/1024=1024MBdata = np.random.uniform([128, 4, 512, 512])data = pickle.dumps(data)try:sock.send(data)except Timeout:# 如果对面端提前关了，那么这里会超时，这里直接跳出breakif random.random() < 0.5:try:r = sock.recv(block=False)print(r)except TryAgain:passprint('end1')sock.close()def proc_recv_data():'''pair端想发就发，想收就收:return:'''sock = nng.Pair0(listen='tcp://127.0.0.1:8855', recv_timeout=500, send_timeout=500)for i in range(10000):try:r = sock.recv(block=False)data = pickle.loads(r)print(i, np.mean(data))except TryAgain:# 如果没有消息passif random.random() < 0.5:try:sock.send(bytes([1]))except Timeout:# 如果对面端提前关了，那么这里会超时，这里直接跳出breakprint('end2')sock.close()if __name__ == '__main__':p1 = Process(target=proc_recv_data, daemon=True)p2 = Process(target=proc_send_data, daemon=True)p1.start()p2.start()p1.join()p2.join()print('success')

_try_pub_sub.py

from pynng import nng from multiprocessing import Process import numpy as np import pickle from pynng.exceptions import TryAgaindef proc_send_data():'''pub端 和push端相似，只管发数据:return:'''sock = nng.Pub0(listen='tcp://127.0.0.1:8855')for i in range(10000):data = np.random.uniform([128, 4, 512, 512])data = pickle.dumps(data)sock.send(data)sock.close()def proc_recv_data():'''sub端 和pull端相似，只管收数据。可以设定订阅标题和取消订阅标题。首次使用时，要设定为接收空标题，不然啥也收不到:return:'''sock = nng.Sub0(dial='tcp://127.0.0.1:8855')# 设定接受空标题sock.subscribe('')for i in range(10000):try:r = sock.recv(block=False)data = pickle.loads(r)print(i, np.mean(data))except TryAgain:print()sock.close()if __name__ == '__main__':p1 = Process(target=proc_recv_data, daemon=True)p2 = Process(target=proc_send_data, daemon=True)p1.start()p2.start()p1.join()p2.join()print('success')

_try_rep_req.py

from pynng import nng from multiprocessing import Process import numpy as np import pickledef proc_send_data():'''req端一开始只能发送数据，并且发送完数据后，必定要接收数据:return:'''sock = nng.Req0(dial='tcp://127.0.0.1:8855')for i in range(10000):data = np.random.uniform([128, 4, 512, 512])data = pickle.dumps(data)sock.send(data)sock.recv()sock.close()def proc_recv_data():'''rep端一开始只能接收数据，并且接收完数据后，必定要发送一个长度不为0的数据:return:'''sock = nng.Rep0(listen='tcp://127.0.0.1:8855')for i in range(10000):r = sock.recv(block=True)data = pickle.loads(r)print(i, np.mean(data))# rep发送的数据长度不能为0sock.send(bytes([0]))sock.close()if __name__ == '__main__':p1 = Process(target=proc_recv_data, daemon=True)p2 = Process(target=proc_send_data, daemon=True)p1.start()p2.start()p1.join()p2.join()print('success')

总结

以上是生活随笔为你收集整理的pynng 超快速上手，但不保熟的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。