Java使用nanomsg消息中间件
邢凌
一，引用的包
<dependency>
<groupId>jnanomsg</groupId>
<artifactId>jnanomsg</artifactId>
<version>0.4.3</version>
</dependency>二，nanomsg的使用，需要针对不同的平台编译并得到动态库，然后引入项目。当前自己在window10-x86-64 和 linux-x86-64的环境下编译了两个动态库。(本次使用的是nanomsg5.1.0)
windows：直接将nanomsg.dll和nanomsg.lib置于类路径下
linux：在类路径下创建目录 linux-x86-64，并将libnanomsg.so文件放入其下
下载路径：https://pan.baidu.com/s/18-wE_QMeYDkwi4kYQfrAaw
三，借用别人的代码
1，PAIR 模式
public class Pair {
private static String url = "tcp://127.0.0.1:7789";
public static void main(String[] args) {
node0();
node1();
}
private static void node0() {
PairSocket socket = new PairSocket();
socket.connect(url);
send(socket);
recv(socket, "node0");
}
private static void node1() {
PairSocket socket = new PairSocket();
socket.bind(url);
send(socket);
recv(socket, "node1");
}
private static void recv(final PairSocket socket, final String nodeName) {
socket.setRecvTimeout(2000); // 设置执行recv的超时时间
new Thread(new Runnable() {
public void run() {
while (true) {
try {
System.out.println(nodeName + ":" + socket.recvString()); // 阻塞socket，直到超时或者有响应
Thread.sleep(1000);
} catch (IOException e) { // 忽略超时
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
}
private static void send(final PairSocket socket) {
socket.setSendTimeout(1100); // 设置执行send的超时时间
new Thread(new Runnable() {
public void run() {
while (true) {
try {
socket.send("hello");
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}).start();
}
}2，Pipeline模式
public class Pipeline {
private static String url = "tcp://127.0.0.1:7789";
public static void main(String[] args) {
node1();
node0();
}
private static void node1() {
final PullSocket socket = new PullSocket();
socket.bind(url);
new Thread(new Runnable() {
public void run() {
while (true) {
try {
System.out.println(socket.recvString()); // 阻塞socket，直到超时或者有响应
Thread.sleep(1000);
} catch (IOException e) { // 忽略超时
// e.printStackTrace();
} catch (InterruptedException e) {
// e.printStackTrace();
}
}
}
}).start();
}
private static void node0() {
final PushSocket socket = new PushSocket();
socket.connect(url);
socket.send("hello");
}
}3，ReqRep模式
public class ReqRep {
private static String url = "tcp://127.0.0.1:7789";
public static void main(String[] args) {
node1();
node0();
}
private static void node1() {
final RepSocket socket = new RepSocket();
socket.bind(url);
new Thread(new Runnable() {
public void run() {
while (true) {
try {
System.out.println( "node1:" + socket.recvString()); // 阻塞socket，直到超时或者有响应
Thread.sleep(1000);
socket.send("world");
} catch (IOException e) { // 忽略超时
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
}
private static void node0() {
final ReqSocket socket = new ReqSocket();
socket.connect(url);
new Thread(new Runnable() {
public void run() {
while (true) {
try {
socket.send("hello");
Thread.sleep(1000);
System.out.println( "node0:" + socket.recvString()); // 阻塞socket，直到超时或者有响应
} catch (InterruptedException e) {
e.printStackTrace();
} catch (IOException e) {
// e.printStackTrace();
}
}
}
}).start();
}
}4，pubsub模式
public class PubSub {
private static String url = "tcp://127.0.0.1:7789";
public static void main(String[] args) {
service();
client0();
client1();
client2();
}
private static void client0() {
client("client0");
}
private static void client1() {
client("client1");
}
private static void client2() {
client("client2");
}
private static void client(final String name) {
final SubSocket socket = new SubSocket();
socket.connect(url);
socket.subscribe("test"); // jnanomsg中 频道的匹配是匹配recv为^test的消息
new Thread(new Runnable() {
public void run() {
while (true) {
try{
System.out.println(name + ":" + socket.recvString());
}catch (IOException e) { // 忽略超时
//e.printStackTrace();
}
}
}
}).start();
}
private static void service() {
final PubSocket socket = new PubSocket();
socket.bind(url);
new Thread(new Runnable() {
public void run() {
while (true) {
try{
socket.send("test1 msg");
Thread.sleep(2000);
}catch (IOException e) { // 忽略超时
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
}
}5，bus 模式
public class Bus {
private static String url0 = "tcp://127.0.0.1:7780";
private static String url1 = "tcp://127.0.0.1:7781";
private static String url2 = "tcp://127.0.0.1:7782";
private static String url3 = "tcp://127.0.0.1:7783";
public static void main(String[] args) {
BusSocket s0 = node("node0", url0, new String[]{url1, url2, url3});
BusSocket s1 = node("node1", url1, new String[]{url2, url3});
BusSocket s2 = node("node2", url2, new String[]{url3});
BusSocket s3 = node("node3", url3, new String[]{});
s0.send("client0 send a");
s1.send("client1 send a");
s2.send("client2 send a");
s3.send("client3 send a");
}
private static BusSocket node(final String name, String self, String[] other) {
final BusSocket socket = new BusSocket();
socket.bind(self);
for (String s : other){
socket.connect(s);
}
new Thread(new Runnable() {
public void run() {
while (true) {
try {
System.out.println(name + ":"+ socket.recvString()); // 阻塞socket，直到超时或者有响应
Thread.sleep(1);
} catch (IOException e) { // 忽略超时
// e.printStackTrace();
} catch (InterruptedException e) {
// e.printStackTrace();
}
}
}
}).start();
return socket;
// socket.connect();
}
}转载于:https://blog.51cto.com/dengshuangfu/2130285
类似资料：