ZMQ dome

秒速五厘米 2021-09-24 22:32 229阅读 0赞

1，pom

<dependency>
    			<groupId>org.zeromq</groupId>
    			<artifactId>jeromq</artifactId>
    			<version>0.3.6</version>
    		</dependency>

2，接收端代码

import org.zeromq.ZMQ;
    import org.zeromq.ZMQ.Context;
    import org.zeromq.ZMQ.Socket;
    /**
     * zmq接收消息
     * @author gyn
     *
     */
    public class ZmqSub extends Thread {
    
    	private Socket socket;
    
    	public ZmqSub(int port) {
    		Context context = ZMQ.context(1);
    		//ZMQ.SUB
    		socket = context.socket(ZMQ.SUB);
    		socket.connect("tcp://127.0.0.1:"+port);//设置ip port
    		socket.subscribe("".getBytes());//设置接收格式
    	}
    
    	@Override
    	public void run() {
    		int i=0;
    		while (true) {
    			if(i==0) {
    				System.err.println("等待收到消息：");
    			}
    			try {
    				byte[] content = socket.recv(0);
    				String string = new String(content);
    				System.err.println("收到消息：" + string);
    				i++;
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

3，发送端代码

import java.util.concurrent.LinkedTransferQueue;
    
    import org.zeromq.ZMQ;
    import org.zeromq.ZMQ.Context;
    import org.zeromq.ZMQ.Socket;
    
    /**
     * zmq发送消息
     * @author gyn
     *
     */
    public class ZmqPub extends Thread {
    
    	private boolean state = true;
    	private Context context;
    	private Socket socket;
    	private LinkedTransferQueue<String> queue = new LinkedTransferQueue<>();
    
    	public ZmqPub(int port) {
    		context = ZMQ.context(1);
    		//单向
    		socket = context.socket(ZMQ.PUB);
    		socket.bind("tcp://127.0.0.1:" + port);//设置ip port
    		socket.setSndHWM(0);
    	}
    	
    	@Override
    	public void run() {
    		while (state) {
    			execute();
    		}
    	}
    
    	public void addTask(String task) {
    		queue.add(task);
    	}
    
    	private void execute() {
    		try {
    			String msg = queue.take();
    			if (msg != null) {
    				sendMessage(msg);
    			}
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    	private void sendMessage(String msg) {
    		try {
    			System.err.println(">>>>>>>>>>>>>>>>>>>>>>>>>>发送消息：" + msg);
    			socket.send(msg.getBytes());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    }

4，main

import java.util.UUID;
    
    public class ZMQTest {
    	public static void main(String[] args) throws InterruptedException {
    		int port=15666;
    		//接收端
    		new ZmqSub(port).start();
    		
    		//发送端
    		ZmqPub zpb=new ZmqPub(port);
    		zpb.start();
    		for (int i = 0; i < 100; i++) {
    			Thread.sleep(1000);
    			zpb.addTask(UUID.randomUUID().toString().replace("-", ""));
    		}
    		
    	}
    }

5，运行main看效果