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

ZeroMQ实战：用Java玩转PUB/SUB和REQ/REP，构建你的第一个分布式温度监控Demo

news 2026/4/23 14:11:19

ZeroMQ实战：用Java构建分布式温度监控系统

想象一下，你正在开发一个智能温室系统，需要实时监控分布在各个角落的温度传感器数据。传统做法可能是让每个传感器直接连接数据库，但这种架构在传感器数量增加时会遇到性能瓶颈。这时，ZeroMQ的消息队列模式就能大显身手——它能让数据像报纸一样被"发布"，而监控中心只需"订阅"自己关心的内容，无需知道数据从哪里来。

1. 环境准备与ZeroMQ基础

在开始编码前，我们需要准备好开发环境。推荐使用Java 11或更高版本，配合Maven进行依赖管理。ZeroMQ的Java绑定有多个实现，这里我们选择活跃度较高的jeromq：

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

ZeroMQ的核心概念可以概括为三个要素：

Context：相当于通信的运行时环境，通常一个应用只需要一个
Socket：通信端点，支持多种模式（PUB/SUB、REQ/REP等）
Transport：通信协议，常用的是TCP和inproc（进程内）

重要提示：ZeroMQ的Socket与传统网络Socket不同，它提供了更高层次的抽象。比如PUB socket会自动处理连接断开和重连，SUB socket可以过滤只接收特定前缀的消息。

2. 温度传感器实现（PUB模式）

我们的温度传感器需要定期发布读数，采用PUB模式最合适。下面是一个带有模拟数据生成功能的实现：

import org.zeromq.ZMQ; import java.util.Random; import java.time.LocalDateTime; public class TemperatureSensor { private static final String[] LOCATIONS = {"温室A区", "温室B区", "储藏室", "户外"}; public static void main(String[] args) { try (ZMQ.Context context = ZMQ.context(1); ZMQ.Socket publisher = context.socket(ZMQ.PUB)) { publisher.bind("tcp://*:5556"); Random random = new Random(); while (!Thread.currentThread().isInterrupted()) { String location = LOCATIONS[random.nextInt(LOCATIONS.length)]; double temperature = 15 + random.nextDouble() * 20; // 15-35℃范围 String timestamp = LocalDateTime.now().toString(); // 发送位置作为消息主题 publisher.sendMore(location); // 发送实际数据：温度|时间 publisher.send(String.format("%.1f|%s", temperature, timestamp)); System.out.printf("[发布] %s: %.1f℃ @ %s\n", location, temperature, timestamp); Thread.sleep(1000 + random.nextInt(1000)); // 1-2秒间隔 } } catch (Exception e) { e.printStackTrace(); } } }

这个实现有几个关键点：

使用sendMore()方法实现多部分消息，第一部分是位置（主题）
数据格式设计为"温度|时间戳"的简单结构
随机间隔模拟真实传感器的不规律上报

性能考虑：在实际部署中，可以考虑批量发送读数以减少网络开销，但需要权衡实时性要求。

3. 监控中心实现（SUB模式）

监控中心需要订阅所有传感器的数据，并进行简单分析。下面是SUB模式的实现：

import org.zeromq.ZMQ; import java.util.HashMap; import java.util.Map; public class MonitoringCenter { private static final Map<String, Double> maxTemperatures = new HashMap<>(); public static void main(String[] args) { try (ZMQ.Context context = ZMQ.context(1); ZMQ.Socket subscriber = context.socket(ZMQ.SUB)) { subscriber.connect("tcp://localhost:5556"); subscriber.subscribe("".getBytes()); // 订阅所有消息 System.out.println("监控中心启动，等待温度数据..."); while (!Thread.currentThread().isInterrupted()) { String location = subscriber.recvStr(); String[] data = subscriber.recvStr().split("\\|"); double temperature = Double.parseDouble(data[0]); String time = data[1]; // 更新最高温度记录 maxTemperatures.put(location, Math.max(temperature, maxTemperatures.getOrDefault(location, Double.MIN_VALUE))); System.out.printf("[接收] %s: %.1f℃ (历史最高: %.1f℃) @ %s\n", location, temperature, maxTemperatures.get(location), time); } } catch (Exception e) { e.printStackTrace(); } } }

监控中心的核心功能包括：

记录每个位置的最高温度
显示实时数据及历史最高值
支持通配符订阅（空字符串表示订阅所有）

扩展思考：如果想只监控特定区域（如只关注"温室A区"），只需修改subscribe参数：

subscriber.subscribe("温室A区".getBytes());

4. 查询服务实现（REQ/REP模式）

为了让用户能主动查询当前状态，我们增加一个请求-响应服务。监控中心将同时扮演REP角色：

// 在MonitoringCenter的try-with-resources中添加 ZMQ.Socket responder = context.socket(ZMQ.REP); responder.bind("tcp://*:5557"); // 新线程处理查询请求 new Thread(() -> { while (!Thread.currentThread().isInterrupted()) { String query = responder.recvStr(); String response; if ("max".equals(query)) { response = maxTemperatures.toString(); } else if ("all".equals(query)) { response = "未知命令，支持: max, all"; } responder.send(response); } }).start();

查询客户端实现：

import org.zeromq.ZMQ; public class QueryClient { public static void main(String[] args) { try (ZMQ.Context context = ZMQ.context(1); ZMQ.Socket requester = context.socket(ZMQ.REQ)) { requester.connect("tcp://localhost:5557"); // 查询最高温度 requester.send("max"); String reply = requester.recvStr(); System.out.println("当前各区域最高温度: " + reply); // 可以继续发送其他查询... } } }

这个设计实现了：

多协议共存：监控中心同时处理SUB和REP socket
线程安全：使用ConcurrentHashMap代替普通HashMap
可扩展的查询协议

5. 系统优化与错误处理

真实环境中需要考虑更多因素，以下是一些改进建议：

心跳检测（防止僵尸连接）：

// 在PUB端设置心跳 socket.setHWM(1000); socket.setHeartbeatIvl(1000); socket.setHeartbeatTimeout(3000);

消息序列化（复杂数据结构）：

// 使用JSON ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(sensorData); socket.send(json); // 接收方 SensorData data = mapper.readValue(receivedStr, SensorData.class);

错误处理表格：