当前位置：首页 > news >正文

Java+nanomsg快速实现去broker的数据通信

news 2025/8/25 16:39:56

先说一下nanomsg+Java需要做什么：
1、nanomsg的so文件的制作与放置路径
2、Java代码引入nanomsg的依赖
3、支持Socket参数的调节（包括ipv4/ipv6的网络支持）

在我目前的认知范围内要与一个通讯目标实现数据交互通常有这些方式
1、broker中间人转发，
例如：kafka，mqtt 等等mq

2、直接连接进行数据交互
例如：tcp，udp
衍生开来就包含了很多http，netty自定义开发，grpc，等等当然也包含了今天的主角nanomsg；

nanomsg官网：https://niwi.nz/jnanomsg/latest/#_introduction
nanomsg用来做什么看看官网对它的介绍，模式挺多的；

我来简单讲一下怎么应用起来并且让程序跑起来：

一、so文件的制作与放置路径

git clone https://github.com/nanomsg/nanomsg.git
cd nanomsg
mkdir build
cd build
cmake ..
make

此时，需要将支持我们系统架构的库文件挪移到对应的文件夹下，以方便底层逻辑的读取！

此时build文件下的：libnanomsg.so(x86下的)就是我们需要的库文件了
注意：在什么样的操作系统下制作出来的就是支持什么系统的文件，比如mac电脑的文件就是：libnanomsg.dylib

对应系统的java+nanomsg的库文件的读取路径大概是：
macOS：/Library/Java/Extensions 或 /Users/你的用户名/Library/Java/Extensions
Linux：/usr/lib 或 /usr/local/lib
Windows：C:\Windows\System32

但是在我测试过程中发现，java对于库文件的读取会有些许变动：
mac真实读取位置：/Library/Java/Extensions/darwin/libnanomsg.dylib
linux(x86)真实读取位置：/usr/lib/libnanomsg.so
其它的需要自己测测了

错误引流：
Caused by: java.lang.UnsatisfiedLinkError: Unable to load library ‘nanomsg’: Native library (darwin/libnanomsg.dylib) not found in resource path

二、Java代码引入nanomsg的依赖

maven：(两个应该都可以的，我用的第一个)

        <dependency><groupId>cs</groupId><artifactId>jnanomsg</artifactId><version>0.4.5</version></dependency>
<!--        <dependency>-->
<!--            <groupId>jnanomsg</groupId>-->
<!--            <artifactId>jnanomsg</artifactId>-->
<!--            <version>0.4.3</version>-->
<!--        </dependency>-->

代码演示：

提一下address的地址支持：
ipv4支持address:
tcp://0.0.0.0:5551
tcp://*:5592
tcp://[::]:5559

ipv6支持address：
tcp://*:5592
tcp://[::]:5559
tcp://[v6某一个网络端口的地址]:5593

推送数据：
我演示的是pub/sub模式，也可以理解为去broker的mq
一个server可以有一群client进行连接订阅server的数据
topic底层应该默认使用的是空格进行组织的数据
即：
topic1 data
topic2 data
topic3 data

    private static String address = "tcp://[::]:5557";private Socket socket;public void afterPropertiesSet() throws Exception {try {socket = new PubSocket();socket.setSocketOpt(Nanomsg.SocketOption.NN_IPV4ONLY,0);//支持ipv6&v4socket.bind(address);log.info("NanoMsgRsuMsgServer socket start success.ad:{}", address);} catch (IOException e) {log.error("创建失败:",e);}}public void publish(String topic, String data) {if (socket == null) {log.warn("nanomsg socket 尚未初始化");}socket.send(topic + " " + data);}

订阅数据：

    public static void main(String[] args) throws InterruptedException {Socket socket = new SubSocket();// 连到远端地址socket.connect("tcp://127.0.0.1:5557");socket.subscribe("topic1");//订阅某一类型的数据socket.subscribe("topic2");
        socket.subscribe(" ");//订阅所有while (true) {System.out.println(socket.recvString());}}

测试可用性：
telnet v6/v4地址端口（可以被用来查看该端口是否可以被使用/服务是否启动）
lsof v6/v4地址:端口(可以来看端口启动的基本信息)

三、支持Socket参数的调节（包括ipv4/ipv6的网络支持）

没有设置这个的话，会爆错误：

socket.setSocketOpt(Nanomsg.SocketOption.NN_IPV4ONLY,0);//支持ipv6&v4
Nanomsg.SocketOption静态枚举内支持很多参数调节：
NN_SUB_UNSUBSCRIBE,
NN_SUB_SUBSCRIBE,
NN_SNDFD,
NN_RCVFD,
NN_LINGER,
NN_SNDBUF,
NN_RCVBUF,
NN_RCVMAXSIZE,
NN_SNDTIMEO,
NN_RCVTIMEO,
NN_RECONNECT_IVL,
NN_RECONNECT_IVL_MAX,
NN_SNDPRIO,
NN_RCVPRIO,
NN_IPV4ONLY,
NN_SOCKET_NAME;
可以自己搜索用法