一 ZMQ简介

ZMQ(即ZeroMq)号称是“史上最快的消息队列”,基于c语言开发的,实时流处理sorm的task之间的通信就是用的zeroMQ。它跟RabbitMQ,ActiveMQ之类有着相当本质的区别,ZeroMQ根本就不是一个消息队列服务器,更像是一组底层网络通讯库,对原有的Socket API加上一层封装,使我们操作更简便。如果说rabbitMQ已经近乎是一个小型操作系统,那么ZeroMQ就像是一个嵌入在操作系统内的一个组件,说白了ZeroMQ就是一组jar包,直接嵌入到项目中就可以运行,它不需要一台独立的服务器来承载整个消息系统。ZeroMQ的性能远远高于其它MQ。ZeroMQ对于消息的处理可以说除却请求-应答模式之外,基本就是不关系消息是否丢失,它只管发送。

ZeroMQ关注的不是消息的可靠送达,而是着眼于端到端的发送、接收。它希望的是尽快完成任务,而不介意部分消息的丢失。但这也并不是说他完全没有持久化的功能,ZeroMQ是具有一定的本地持久化的功能的,但是能保存的数据量比较有限,而且是暂存于内存中的。

虽然ZeroMQ在高并发环境下不会出问题,但是有可能会导致本地的缓存区被塞满而导致消息丢失的情况。所以不推荐在并发量较高的情境下使用ZeroMQ.

ZeroMQ与其他MQ类似,也实现了3中最基本的工作模式:发布-订阅,请求-应答,管道模式

二 springboot集成ZeroMq

  1. 添加ZeroMq依赖
<dependency>
   <groupId>org.zeromq</groupId>
    <artifactId>jeromq</artifactId>
    <version>0.5.2</version>
</dependency>
  1. 服务端接收消息
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import org.zeromq.ZMQ;

/**
 * @Author:sgw
 * @Date:2023/12/27
 * @Description: 监听zmq消息
 */
@Component
public class ZmqServer  implements CommandLineRunner {

    @Override
    public void run(String... args) throws Exception {
        try (
               //一般参数写1足够,如果希望创建多个IO线程,不要超出cpu核数
                ZMQ.Context context = ZMQ.context(1);
                ZMQ.Socket socket = context.socket(ZMQ.REP)) {
            // 绑定端口
            socket.bind("tcp://*:5555");
            while (true) {
                // 等待接收消息
                byte[] request = socket.recv(0);
                String text = new String(request);
                System.out.println("接收到消息: " + text);
                // 返回消息
                byte[] reply = "你好,我是服务端。".getBytes();
                socket.send(reply, 0);
            }
        }
    }
}
  1. 客户端发送消息
public class ZmqClient {
    public static void main(String[] args) {
        try (ZMQ.Context context = ZMQ.context(1);
             ZMQ.Socket socket = context.socket(ZMQ.REQ)) {
            // 连接服务端
            socket.connect("tcp://localhost:5555");
            // 发送消息
            String text = "你好,我是客户端。";
            byte[] request = text.getBytes();
            socket.send(request, 0);
            System.out.println("发送消息:" + text);
            // 等待回复
            byte[] reply = socket.recv(0);
            String response = new String(reply);
            System.out.println("接收到回复:" + response);
        }
    }
}
Logo

魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。

更多推荐