ZMQ消息队列 PUSH/PULL PUB/SUB REQ/REP

1.REQ/REP 客户端（Client）/ 服务器（Server）

import org.zeromq.ZContext;
import org.zeromq.ZMQ;

/**
 * @Description: （服务器）
 * @CreateTime: 2024/3/20 18:22
 */
public class Server {
    public static void main(String[] args) throws InterruptedException {

        ZMQ.Context context = ZMQ.context(1);
        ZMQ.Socket responder = context.socket(ZMQ.REP);

        //使服务器端通过tcp协议通信，监听5555端口
        responder.bind("tcp://*:5555");
        while (!Thread.currentThread().isInterrupted()) {
            byte[] request = responder.recv(0);
            System.out.println("Received Hello");
            Thread.sleep(1000);
            String reply = "World";
            responder.send(reply.getBytes(), 0);
        }

        //关闭服务器端的上下文及套接字
        responder.close();
        context.close();
    }
}

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/**
 * @Description: （客户端）
 * @CreateTime: 2024/3/20 18:23
 */
public class Client {
    public static void main(String[] args) {
        //创立客户端的上下文捷套接字
        Context context = ZMQ.context(1);
        System.out.println("Connecting to hello world server…");
        Socket requester = context.socket(ZMQ.REQ);

        //讲客户端绑定在5555端口
        requester.connect("tcp://localhost:5555");
        for (int requestNbr = 0; requestNbr != 100; requestNbr++) {
            String request = "Hello";
            System.out.println("Sending Hello " + requestNbr);
            requester.send(request.getBytes(), 0);
            byte[] reply = requester.recv(0);
            System.out.println("Received " + new String(reply) + " " + requestNbr);
        }
        //关闭客户端的上下文套接字
        requester.close();
        context.term();
    }
}

2.PUSH/PULL 模式: 生产者（Producer）/消费者（Consumer）

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

import java.util.concurrent.atomic.AtomicInteger;

import static org.zeromq.ZMQ.context;

/**
 * @Description: 拉取消息 （服务器）
 * @CreateTime: 2024/3/20 18:29
 */
public class Pull {
    public static void main(String args[]) {
        final AtomicInteger number = new AtomicInteger(0);
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable(){
                private int here = 0;
                public void run() {
                    // TODO Auto-generated method stub
                    Context context = context(1);
                    Socket pull = context.socket(ZMQ.PULL);
                    pull.connect("ipc://fjs");
                    //pull.connect("ipc://fjs");
                    while (true) {
                        String message = new String(pull.recv());
                        int now = number.incrementAndGet();
                        here++;
                        if (now % 1000000 == 0) {
                            System.out.println(now + "  here is : " + here);
                        }
                    }
                }
            }).start();
        }
    }
}

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/**
 * @Description: PUSH 推送消息 （客户端）
 * @CreateTime: 2024/3/20 18:29
 */
public class Push {
    public static void main(String[] args) {
        Context context = ZMQ.context(1);
        Socket push = context.socket(ZMQ.PUSH);
        push.bind("ipc://fjs");
        for (int i = 0; i < 10000000; i++) {
            push.send("hello".getBytes(), i);
        }
        push.close();
        context.term();
    }
}

3.PUB/SUB 模式: 发布者（Publisher）/订阅者（Subscriber）

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/**
 * @Description: 发布者（Publisher） （服务器）
 * @CreateTime: 2024/3/20 18:25
 */
public class ZMQ_PUB {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个 ZeroMQ 上下文
        Context context = ZMQ.context(1);
        // 创建一个发布者套接字
        Socket publisher = context.socket(zmq.ZMQ.ZMQ_PUB);
        publisher.bind("tcp://*:5555");
        Thread.sleep(3000);

//        normalSend(publisher);
        topicSend(publisher);

        // 关闭资源
        context.close();
        publisher.close();
    }

    //普通发送
    public static void normalSend(Socket publisher) throws InterruptedException {
        // 发送消息
        for (int i = 0; i < 100; i++) {
            publisher.send(("admin " + i).getBytes(), ZMQ.NOBLOCK);
            System.out.println("pub msg " + i);
            Thread.sleep(1000);
        }
    }

    //带主题的发送
    public static void topicSend(Socket publisher) throws InterruptedException {
        // 发送消息
        for (int i = 0; i < 10; i++) {
            String topic = "NEWS";
            if (i % 2 == 0) {
                topic = "WEATHER";
            }
            String message = topic + " Message " + i;

            //此处发送了主题字段和消息内容，注意都的 topic 主题设置的开头的
            publisher.send(topic.getBytes(), 0);
            publisher.send(message.getBytes(), 0);

            System.out.println("Sent: [" + message + "]");
            Thread.sleep(1000); // 每秒发送一条消息
        }
    }
}

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/**
 * @Description: 订阅者（Subscriber） （客户端）
 * @CreateTime: 2024/3/20 18:26
 */

public class ZMQ_SUB {
    public static void main(String[] args) {
        // 创建一个 ZeroMQ 上下文
        Context context = ZMQ.context(1);
        // 创建一个订阅者套接字
        Socket subscriber = context.socket(zmq.ZMQ.ZMQ_SUB);
        subscriber.connect("tcp://localhost:5555");
//        normalRecv(context, subscriber);
        topicRecv(subscriber);
    }

    //接受普通信息，设置主日为 ""
    public static void normalRecv(Context context, Socket subscriber) {
        // 订阅所有主题（空字符串表示订阅所有消息）
        subscriber.subscribe("".getBytes());

        // 接收并打印消息
        for (int i = 0; i < 100; i++) {
            //Receive a message.
            String string = new String(subscriber.recv(0));
            System.out.println("recv 1" + string);
        }
        //关闭套接字和上下文
        subscriber.close();
        context.term();
    }

    public static void topicRecv(Socket subscriber) {
        // 订阅特定主题（"NEWS"）
        // 所谓主题，就是字符串的开始字段。 只接受 NEWS 开始的字段
        subscriber.subscribe("NEWS".getBytes());

        // 接收并打印消息
        while (true) {
            byte[] topic = subscriber.recv(0);
            byte[] message = subscriber.recv(0);
            System.out.println("Received: [" + new String(topic) + "][" + new String(message) + "]");

        }
    }

}

        <!-- https://mvnrepository.com/artifact/org.zeromq/jeromq -->
        <dependency>
            <groupId>org.zeromq</groupId>
            <artifactId>jeromq</artifactId>
            <version>0.5.3</version>
        </dependency>

总结
    PUSH/PULL 模式: 生产者（Producer）/消费者（Consumer）
        •用途：生产者-消费者模型。
        •特点：单向通信、无序消息、可靠性。
        •应用场景：任务分发、异步通信。
    PUB/SUB 模式:  发布者（Publisher）/订阅者（Subscriber）
        •用途：发布者-订阅者模型。
        •特点：多播通信、主题过滤、广播模式、无序消息、可靠性。
        •应用场景：实时消息更新、数据分发。
    REQ/REP 模式: 客户端（Client）/ 服务器（Server）
        •用途：请求-响应模型。
        •特点：双向通信、有序消息、可靠通信、阻塞模式。
        •应用场景：远程过程调用（RPC）、命令-响应。
通过这三种模式，ZeroMQ 提供了灵活的消息传递方式，可以根据实际需求选择合适的模式来构建应用程序。