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

从 0 到 1 搭建实时数据看板：RabbitMQ+WebSocket 实战指南

news 2025/10/9 7:06:50

在当今数据驱动的时代，实时数据可视化已成为企业决策的核心需求。无论是电商平台的实时交易监控，还是物联网系统的设备状态看板，都需要高效、稳定的数据实时采集与推送能力。本文将带你构建一套完整的实时数据处理 pipeline，通过 RabbitMQ 实现高效的数据采集，结合 WebSocket 技术将数据毫秒级推送到前端看板，让你轻松掌握实时数据处理的核心技术。

技术选型与架构设计

核心技术栈解析

实现实时数据看板需要解决三个关键问题：数据采集、消息传递和实时推送。我们选择以下技术栈组合：

  • 数据采集层：多样化数据源接入，包括数据库变更、API 接口、日志文件等
  • 消息中间件：RabbitMQ 3.13.0，提供可靠的消息投递和灵活的路由策略
  • 实时推送层：WebSocket，基于 Spring WebSocket 6.1.2 实现双向通信
  • 后端框架：Spring Boot 3.2.0，简化开发流程
  • 前端框架：Vue 3 + ECharts 5.4.3，实现数据可视化
  • 数据库：MySQL 8.0.35，存储元数据和历史数据
  • ORM 框架：MyBatis-Plus 3.5.5，简化数据库操作
  • JSON 处理：FastJSON2 2.0.45，高效 JSON 序列化 / 反序列化

整体架构设计

下面是系统的整体架构图，清晰展示了数据从产生到最终展示的完整流程：

数据流转流程

  1. 多源数据通过采集器接入系统，统一格式后发送到 RabbitMQ 交换机
  2. 交换机根据路由规则将数据分发到不同队列：
    • 正常数据进入数据处理队列
    • 异常数据进入死信队列等待后续处理
  3. 应用服务消费数据处理队列中的消息，进行业务处理
  4. 处理后的数据一方面持久化到数据库，另一方面通过 WebSocket 推送到前端
  5. 前端接收数据后，通过 ECharts 实时更新可视化图表

环境搭建与项目初始化

开发环境准备

首先确保你的开发环境满足以下要求：

  • JDK 17+（推荐 Amazon Corretto 17.0.10）
  • Maven 3.9.6
  • MySQL 8.0.35+
  • RabbitMQ 3.13.0（带 Management 插件）
  • Node.js 18.19.0+（用于前端开发）
  • IDE：IntelliJ IDEA 2023.3+

RabbitMQ 安装与配置

  1. 安装 RabbitMQ（以 Linux 为例）：
# 安装Erlang依赖
sudo apt update
sudo apt install erlang# 安装RabbitMQ
sudo apt install rabbitmq-server# 启动服务
sudo systemctl start rabbitmq-server# 启用管理插件
sudo rabbitmq-plugins enable rabbitmq_management# 配置管理员用户
sudo rabbitmqctl add_user admin password
sudo rabbitmqctl set_user_tags admin administrator
sudo rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

  1. 访问 RabbitMQ 管理界面：http://localhost:15672，使用账号 admin/password 登录

  2. 创建所需的交换机和队列：

    • 交换机：data.collect.exchange（类型：topic）
    • 数据处理队列：data.process.queue，绑定键：data.*
    • 异常队列：data.error.queue，绑定键：error.*

项目初始化（Spring Boot 后端）

  1. 创建 Maven 项目，pom.xml 配置如下：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.2.0</version><relativePath/></parent><groupId>com.realtime</groupId><artifactId>data-dashboard</artifactId><version>0.0.1-SNAPSHOT</version><name>data-dashboard</name><description>实时数据看板系统</description><properties><java.version>17</java.version><mybatis-plus.version>3.5.5</mybatis-plus.version><fastjson2.version>2.0.45</fastjson2.version><lombok.version>1.18.30</lombok.version><swagger.version>3.0.0</swagger.version></properties><dependencies><!-- Spring Boot核心 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency><!-- 数据库 --><dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><scope>runtime</scope></dependency><!-- MyBatis-Plus --><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>${mybatis-plus.version}</version></dependency><!-- JSON处理 --><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>${fastjson2.version}</version></dependency><!-- 工具类 --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><scope>provided</scope></dependency><dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>33.1.0-jre</version></dependency><!-- API文档 --><dependency><groupId>org.springdoc</groupId><artifactId>springdoc-openapi-starter-webmvc-ui</artifactId><version>2.2.0</version></dependency><!-- 测试 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build>
</project>
  1. 配置文件（application.yml）：
spring:application:name: data-dashboarddatasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/realtime_dashboard?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghaiusername: rootpassword: rootrabbitmq:host: localhostport: 5672username: adminpassword: passwordvirtual-host: /listener:simple:acknowledge-mode: manual  # 手动确认消息concurrency: 3  # 消费者并发数max-concurrency: 10  # 最大消费者并发数prefetch: 100  # 每次从队列获取的消息数template:retry:enabled: true  # 启用重试max-attempts: 3  # 最大重试次数initial-interval: 1000ms  # 初始重试间隔server:port: 8080servlet:context-path: /apimybatis-plus:mapper-locations: classpath*:/mapper/**/*.xmlglobal-config:db-config:id-type: autologic-delete-field: deletedlogic-delete-value: 1logic-not-delete-value: 0configuration:map-underscore-to-camel-case: truelog-impl: org.apache.ibatis.logging.stdout.StdOutImplspringdoc:api-docs:path: /api-docsswagger-ui:path: /swagger-ui.htmloperationsSorter: methodlogging:level:com.realtime: infoorg.springframework.amqp.rabbit.listener: warn
  1. 数据库初始化脚本（MySQL）：
CREATE DATABASE IF NOT EXISTS realtime_dashboard DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE realtime_dashboard;-- 数据采集记录表
CREATE TABLE IF NOT EXISTS data_collection (id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键ID',data_type VARCHAR(50) NOT NULL COMMENT '数据类型',data_content JSON NOT NULL COMMENT '数据内容',source VARCHAR(100) NOT NULL COMMENT '数据来源',status TINYINT NOT NULL DEFAULT 0 COMMENT '状态：0-新采集 1-已处理 2-异常',create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',process_time DATETIME NULL COMMENT '处理时间',deleted TINYINT NOT NULL DEFAULT 0 COMMENT '逻辑删除：0-未删除 1-已删除',INDEX idx_create_time (create_time),INDEX idx_data_type (data_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据采集记录表';-- 设备状态表（示例）
CREATE TABLE IF NOT EXISTS device_status (id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键ID',device_id VARCHAR(50) NOT NULL COMMENT '设备ID',device_name VARCHAR(100) NOT NULL COMMENT '设备名称',status TINYINT NOT NULL COMMENT '状态：0-离线 1-在线 2-异常',temperature DECIMAL(5,2) NULL COMMENT '温度',humidity DECIMAL(5,2) NULL COMMENT '湿度',last_update_time DATETIME NOT NULL COMMENT '最后更新时间',deleted TINYINT NOT NULL DEFAULT 0 COMMENT '逻辑删除：0-未删除 1-已删除',UNIQUE KEY uk_device_id (device_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='设备状态表';-- 系统配置表
CREATE TABLE IF NOT EXISTS system_config (id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键ID',config_key VARCHAR(50) NOT NULL COMMENT '配置键',config_value VARCHAR(255) NOT NULL COMMENT '配置值',config_desc VARCHAR(255) NULL COMMENT '配置描述',update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',deleted TINYINT NOT NULL DEFAULT 0 COMMENT '逻辑删除：0-未删除 1-已删除',UNIQUE KEY uk_config_key (config_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='系统配置表';

核心数据模型设计

良好的数据模型设计是系统稳定运行的基础，我们需要设计一套灵活且可扩展的数据结构来应对不同类型的实时数据。

通用数据传输对象（DTO）

首先定义一个通用的数据传输对象，作为所有实时数据的载体：

package com.realtime.dto;import com.alibaba.fastjson2.annotation.JSONField;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.time.LocalDateTime;/*** 通用数据传输对象，用于在系统各组件间传递数据** @author ken*/
@Data
@Schema(description = "通用数据传输对象")
public class GenericDataDTO {/*** 数据唯一标识*/@Schema(description = "数据唯一标识")private String dataId;/*** 数据类型（用于路由和处理）*/@Schema(description = "数据类型", example = "device.status, order.payment")private String dataType;/*** 数据来源*/@Schema(description = "数据来源", example = "iot-gateway, order-service")private String source;/*** 数据内容（JSON格式）*/@Schema(description = "数据内容（JSON格式）")private String content;/*** 数据产生时间*/@Schema(description = "数据产生时间")@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")@JSONField(format = "yyyy-MM-dd HH:mm:ss")private LocalDateTime timestamp;
}

设备状态数据模型

以设备状态数据为例，展示具体业务数据模型的设计：

package com.realtime.dto;import com.alibaba.fastjson2.annotation.JSONField;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.time.LocalDateTime;/*** 设备状态数据DTO** @author ken*/
@Data
@Schema(description = "设备状态数据DTO")
public class DeviceStatusDTO {/*** 设备ID*/@Schema(description = "设备ID", example = "device-1001")private String deviceId;/*** 设备名称*/@Schema(description = "设备名称", example = "温度传感器-01")private String deviceName;/*** 设备状态：0-离线 1-在线 2-异常*/@Schema(description = "设备状态：0-离线 1-在线 2-异常", example = "1")private Integer status;/*** 温度（摄氏度）*/@Schema(description = "温度（摄氏度）", example = "25.6")private Double temperature;/*** 湿度（百分比）*/@Schema(description = "湿度（百分比）", example = "45.2")private Double humidity;/*** 信号强度*/@Schema(description = "信号强度", example = "95")private Integer signalStrength;/*** 数据采集时间*/@Schema(description = "数据采集时间")@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")@JSONField(format = "yyyy-MM-dd HH:mm:ss")private LocalDateTime collectTime;
}

实体类设计（MyBatis-Plus）

对应数据库表的实体类设计：

package com.realtime.entity;import com.baomidou.mybatisplus.annotation.*;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.time.LocalDateTime;/*** 数据采集记录表实体类** @author ken*/
@Data
@TableName("data_collection")
@Schema(description = "数据采集记录表")
public class DataCollection {/*** 主键ID*/@TableId(type = IdType.AUTO)@Schema(description = "主键ID")private Long id;/*** 数据类型*/@Schema(description = "数据类型")private String dataType;/*** 数据内容（JSON格式）*/@Schema(description = "数据内容（JSON格式）")private String dataContent;/*** 数据来源*/@Schema(description = "数据来源")private String source;/*** 状态：0-新采集 1-已处理 2-异常*/@Schema(description = "状态：0-新采集 1-已处理 2-异常")private Integer status;/*** 创建时间*/@TableField(fill = FieldFill.INSERT)@Schema(description = "创建时间")private LocalDateTime createTime;/*** 处理时间*/@Schema(description = "处理时间")private LocalDateTime processTime;/*** 逻辑删除：0-未删除 1-已删除*/@TableLogic@Schema(description = "逻辑删除：0-未删除 1-已删除")private Integer deleted;
}

package com.realtime.entity;import com.baomidou.mybatisplus.annotation.*;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.math.BigDecimal;
import java.time.LocalDateTime;/*** 设备状态表实体类** @author ken*/
@Data
@TableName("device_status")
@Schema(description = "设备状态表")
public class DeviceStatus {/*** 主键ID*/@TableId(type = IdType.AUTO)@Schema(description = "主键ID")private Long id;/*** 设备ID*/@Schema(description = "设备ID")private String deviceId;/*** 设备名称*/@Schema(description = "设备名称")private String deviceName;/*** 状态：0-离线 1-在线 2-异常*/@Schema(description = "状态：0-离线 1-在线 2-异常")private Integer status;/*** 温度*/@Schema(description = "温度")private BigDecimal temperature;/*** 湿度*/@Schema(description = "湿度")private BigDecimal humidity;/*** 最后更新时间*/@TableField(fill = FieldFill.INSERT_UPDATE)@Schema(description = "最后更新时间")private LocalDateTime lastUpdateTime;/*** 逻辑删除：0-未删除 1-已删除*/@TableLogic@Schema(description = "逻辑删除：0-未删除 1-已删除")private Integer deleted;
}

RabbitMQ 消息处理模块

RabbitMQ 作为系统的消息中枢，负责接收、路由和分发所有实时数据。我们需要设计合理的消息生产者、消费者以及消息处理机制。

RabbitMQ 配置类

首先配置 RabbitMQ 的交换机、队列和绑定关系：

package com.realtime.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** RabbitMQ配置类，定义交换机、队列及绑定关系** @author ken*/
@Configuration
public class RabbitMQConfig {/*** 数据采集交换机名称*/public static final String DATA_COLLECT_EXCHANGE = "data.collect.exchange";/*** 数据处理队列名称*/public static final String DATA_PROCESS_QUEUE = "data.process.queue";/*** 异常数据队列名称*/public static final String DATA_ERROR_QUEUE = "data.error.queue";/*** 数据路由键前缀*/public static final String DATA_ROUTING_KEY_PREFIX = "data.";/*** 异常路由键前缀*/public static final String ERROR_ROUTING_KEY_PREFIX = "error.";/*** 创建数据采集交换机（topic类型）** @return 交换机实例*/@Beanpublic TopicExchange dataCollectExchange() {// durable: true 持久化交换机，重启RabbitMQ后依然存在// autoDelete: false 不自动删除，当没有绑定关系时也不删除return ExchangeBuilder.topicExchange(DATA_COLLECT_EXCHANGE).durable(true).autoDelete(false).build();}/*** 创建数据处理队列** @return 队列实例*/@Beanpublic Queue dataProcessQueue() {// durable: true 持久化队列// exclusive: false 非排他队列// autoDelete: false 不自动删除return QueueBuilder.durable(DATA_PROCESS_QUEUE).exclusive(false).autoDelete(false).build();}/*** 创建异常数据队列** @return 队列实例*/@Beanpublic Queue dataErrorQueue() {return QueueBuilder.durable(DATA_ERROR_QUEUE).exclusive(false).autoDelete(false).build();}/*** 绑定数据处理队列到交换机** @param dataProcessQueue 数据处理队列* @param dataCollectExchange 数据采集交换机* @return 绑定关系*/@Beanpublic Binding dataProcessBinding(Queue dataProcessQueue, TopicExchange dataCollectExchange) {// 路由键模式：data.* 匹配所有以data.开头的二级路由键return BindingBuilder.bind(dataProcessQueue).to(dataCollectExchange).with(DATA_ROUTING_KEY_PREFIX + "*");}/*** 绑定异常数据队列到交换机** @param dataErrorQueue 异常数据队列* @param dataCollectExchange 数据采集交换机* @return 绑定关系*/@Beanpublic Binding dataErrorBinding(Queue dataErrorQueue, TopicExchange dataCollectExchange) {// 路由键模式：error.* 匹配所有以error.开头的二级路由键return BindingBuilder.bind(dataErrorQueue).to(dataCollectExchange).with(ERROR_ROUTING_KEY_PREFIX + "*");}
}

消息生产者

实现一个通用的消息生产者，用于将不同类型的数据发送到 RabbitMQ：

package com.realtime.producer;import com.alibaba.fastjson2.JSON;
import com.realtime.config.RabbitMQConfig;
import com.realtime.dto.GenericDataDTO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/*** 数据消息生产者，负责将数据发送到RabbitMQ** @author ken*/
@Slf4j
@Component
@RequiredArgsConstructor
public class DataMessageProducer {private final RabbitTemplate rabbitTemplate;/*** 发送数据消息到RabbitMQ** @param data 通用数据传输对象*/public void sendDataMessage(GenericDataDTO data) {if (ObjectUtils.isEmpty(data)) {log.warn("发送的数据消息为空，不进行处理");return;}if (StringUtils.isEmpty(data.getDataType())) {log.error("数据类型为空，无法发送消息: {}", JSON.toJSONString(data));return;}try {// 构建路由键：data.{dataType}String routingKey = RabbitMQConfig.DATA_ROUTING_KEY_PREFIX + data.getDataType();// 发送消息rabbitTemplate.convertAndSend(RabbitMQConfig.DATA_COLLECT_EXCHANGE,routingKey,JSON.toJSONString(data));log.info("数据消息发送成功，dataId: {}, routingKey: {}", data.getDataId(), routingKey);} catch (Exception e) {log.error("数据消息发送失败，data: {}", JSON.toJSONString(data), e);throw new RuntimeException("发送数据消息到RabbitMQ失败", e);}}/*** 发送异常数据消息到RabbitMQ** @param data 通用数据传输对象* @param errorMsg 错误信息*/public void sendErrorDataMessage(GenericDataDTO data, String errorMsg) {if (ObjectUtils.isEmpty(data)) {log.warn("发送的异常数据消息为空，不进行处理");return;}try {// 构建路由键：error.{dataType}String routingKey = RabbitMQConfig.ERROR_ROUTING_KEY_PREFIX + (StringUtils.hasText(data.getDataType()) ? data.getDataType() : "unknown");// 构建异常消息内容String errorContent = String.format("数据处理异常: %s, 原始数据: %s", errorMsg, JSON.toJSONString(data));// 发送消息rabbitTemplate.convertAndSend(RabbitMQConfig.DATA_COLLECT_EXCHANGE,routingKey,errorContent);log.warn("异常数据消息发送成功，dataId: {}, error: {}", data.getDataId(), errorMsg);} catch (Exception e) {log.error("异常数据消息发送失败，data: {}", JSON.toJSONString(data), e);}}
}

数据采集器示例

实现一个模拟设备数据采集器，定期生成设备状态数据并发送到 RabbitMQ：

package com.realtime.collector;import com.alibaba.fastjson2.JSON;
import com.realtime.dto.DeviceStatusDTO;
import com.realtime.dto.GenericDataDTO;
import com.realtime.producer.DataMessageProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;/*** 设备数据采集器，模拟采集设备状态数据** @author ken*/
@Slf4j
@Component
@RequiredArgsConstructor
public class DeviceDataCollector {private final DataMessageProducer dataMessageProducer;// 模拟设备列表private static final String[] DEVICE_IDS = {"device-1001", "device-1002", "device-1003", "device-1004", "device-1005"};private static final String[] DEVICE_NAMES = {"温度传感器-01", "湿度传感器-01", "压力传感器-01", "振动传感器-01", "液位传感器-01"};/*** 定时采集设备数据（每5秒执行一次）*/@Scheduled(fixedRate = 5000)public void collectDeviceData() {log.info("开始采集设备数据...");// 为每个设备生成一条状态数据for (int i = 0; i < DEVICE_IDS.length; i++) {DeviceStatusDTO deviceStatus = generateDeviceStatus(i);sendDeviceData(deviceStatus);}log.info("设备数据采集完成");}/*** 生成模拟设备状态数据** @param index 设备索引* @return 设备状态数据*/private DeviceStatusDTO generateDeviceStatus(int index) {ThreadLocalRandom random = ThreadLocalRandom.current();DeviceStatusDTO status = new DeviceStatusDTO();status.setDeviceId(DEVICE_IDS[index]);status.setDeviceName(DEVICE_NAMES[index]);// 随机生成状态：80%概率在线，15%概率离线，5%概率异常int statusCode = random.nextInt(100);if (statusCode < 80) {status.setStatus(1); // 在线} else if (statusCode < 95) {status.setStatus(0); // 离线} else {status.setStatus(2); // 异常}// 只有在线状态才生成温度和湿度数据if (status.getStatus() == 1) {// 温度：15-35度之间status.setTemperature(15 + random.nextDouble() * 20);// 湿度：30-70%之间status.setHumidity(30 + random.nextDouble() * 40);// 信号强度：50-100之间status.setSignalStrength(50 + random.nextInt(51));}status.setCollectTime(LocalDateTime.now());return status;}/*** 将设备数据发送到RabbitMQ** @param deviceStatus 设备状态数据*/private void sendDeviceData(DeviceStatusDTO deviceStatus) {GenericDataDTO data = new GenericDataDTO();data.setDataId(UUID.randomUUID().toString());data.setDataType("device.status");data.setSource("device-collector");data.setContent(JSON.toJSONString(deviceStatus));data.setTimestamp(LocalDateTime.now());dataMessageProducer.sendDataMessage(data);}
}

消息消费者

实现消息消费者，从 RabbitMQ 接收消息并进行处理：

package com.realtime.consumer;import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.Channel;
import com.realtime.dto.GenericDataDTO;
import com.realtime.service.DataProcessService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;import java.io.IOException;/*** 数据消息消费者，从RabbitMQ接收数据并处理** @author ken*/
@Slf4j
@Component
@RequiredArgsConstructor
public class DataMessageConsumer {private final DataProcessService dataProcessService;/*** 消费数据处理队列中的消息** @param message 消息对象* @param channel 信道*/@RabbitListener(queues = "data.process.queue")public void consumeDataMessage(Message message, Channel channel) {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {String messageBody = new String(message.getBody());log.info("收到数据消息，deliveryTag: {}, 内容: {}", deliveryTag, messageBody);// 解析消息GenericDataDTO data = JSON.parseObject(messageBody, GenericDataDTO.class);if (ObjectUtils.isEmpty(data)) {log.error("消息解析失败，内容为空，deliveryTag: {}", deliveryTag);// 消息解析失败，直接确认并丢弃channel.basicAck(deliveryTag, false);return;}// 处理数据dataProcessService.processData(data);// 手动确认消息channel.basicAck(deliveryTag, false);log.info("数据消息处理完成并确认，dataId: {}, deliveryTag: {}", data.getDataId(), deliveryTag);} catch (Exception e) {log.error("处理数据消息异常，deliveryTag: {}", deliveryTag, e);try {// 处理异常，拒绝消息并将其放入死信队列channel.basicNack(deliveryTag, false, false);log.info("数据消息处理失败，已拒绝并放入死信队列，deliveryTag: {}", deliveryTag);} catch (IOException ex) {log.error("拒绝消息失败，deliveryTag: {}", deliveryTag, ex);}}}/*** 消费异常数据队列中的消息** @param message 消息对象* @param channel 信道*/@RabbitListener(queues = "data.error.queue")public void consumeErrorDataMessage(Message message, Channel channel) {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {String errorMessage = new String(message.getBody());log.error("收到异常数据消息，deliveryTag: {}, 内容: {}", deliveryTag, errorMessage);// 这里可以添加异常数据的处理逻辑，如保存到数据库、发送告警等// 手动确认消息channel.basicAck(deliveryTag, false);} catch (Exception e) {log.error("处理异常数据消息失败，deliveryTag: {}", deliveryTag, e);try {channel.basicNack(deliveryTag, false, false);} catch (IOException ex) {log.error("拒绝异常消息失败，deliveryTag: {}", deliveryTag, ex);}}}
}

数据处理服务实现

数据处理服务是连接消息消费和 WebSocket 推送的中间环节，负责数据的解析、存储和转发。

数据处理接口与实现

首先定义数据处理接口：

package com.realtime.service;import com.realtime.dto.GenericDataDTO;/*** 数据处理服务接口** @author ken*/
public interface DataProcessService {/*** 处理接收到的数据** @param data 通用数据传输对象*/void processData(GenericDataDTO data);
}

实现数据处理服务：

package com.realtime.service.impl;import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.realtime.dto.DeviceStatusDTO;
import com.realtime.dto.GenericDataDTO;
import com.realtime.entity.DataCollection;
import com.realtime.entity.DeviceStatus;
import com.realtime.mapper.DataCollectionMapper;
import com.realtime.mapper.DeviceStatusMapper;
import com.realtime.producer.DataMessageProducer;
import com.realtime.service.DataProcessService;
import com.realtime.service.WebSocketService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Objects;/*** 数据处理服务实现类** @author ken*/
@Slf4j
@Service
@RequiredArgsConstructor
public class DataProcessServiceImpl implements DataProcessService {private final DataCollectionMapper dataCollectionMapper;private final DeviceStatusMapper deviceStatusMapper;private final DataMessageProducer dataMessageProducer;private final WebSocketService webSocketService;/*** 处理接收到的数据** @param data 通用数据传输对象*/@Override@Transactional(rollbackFor = Exception.class)public void processData(GenericDataDTO data) {if (ObjectUtils.isEmpty(data)) {log.warn("处理的数据为空，不进行操作");return;}log.info("开始处理数据，dataId: {}, dataType: {}", data.getDataId(), data.getDataType());try {// 1. 保存原始数据到数据库saveRawData(data);// 2. 根据数据类型进行不同处理switch (data.getDataType()) {case "device.status":processDeviceStatusData(data);break;case "order.payment":processOrderPaymentData(data);break;// 可以添加更多数据类型的处理逻辑default:log.info("未定义的数据类型，仅保存原始数据，dataType: {}", data.getDataType());// 推送原始数据到前端webSocketService.broadcast(data);}// 3. 更新数据状态为已处理updateDataStatus(data, 1);log.info("数据处理完成，dataId: {}", data.getDataId());} catch (Exception e) {log.error("处理数据异常，dataId: {}", data.getDataId(), e);// 更新数据状态为异常updateDataStatus(data, 2);// 发送异常数据到错误队列dataMessageProducer.sendErrorDataMessage(data, e.getMessage());// 抛出异常，触发事务回滚throw new RuntimeException("处理数据失败: " + e.getMessage(), e);}}/*** 保存原始数据到数据库** @param data 通用数据传输对象*/private void saveRawData(GenericDataDTO data) {DataCollection collection = new DataCollection();collection.setDataType(data.getDataType());collection.setDataContent(data.getContent());collection.setSource(data.getSource());collection.setStatus(0); // 0-新采集collection.setCreateTime(LocalDateTime.now());int rows = dataCollectionMapper.insert(collection);if (rows <= 0) {throw new RuntimeException("保存原始数据失败");}log.info("原始数据保存成功，dataId: {}, dbId: {}", data.getDataId(), collection.getId());}/*** 处理设备状态数据** @param data 通用数据传输对象*/private void processDeviceStatusData(GenericDataDTO data) {if (!StringUtils.hasText(data.getContent())) {throw new RuntimeException("设备状态数据内容为空");}// 解析设备状态数据DeviceStatusDTO deviceStatusDTO = JSON.parseObject(data.getContent(), DeviceStatusDTO.class);if (ObjectUtils.isEmpty(deviceStatusDTO) || !StringUtils.hasText(deviceStatusDTO.getDeviceId())) {throw new RuntimeException("设备状态数据解析失败，设备ID为空");}log.info("开始处理设备状态数据，deviceId: {}", deviceStatusDTO.getDeviceId());// 查询设备是否已存在DeviceStatus existingDevice = deviceStatusMapper.selectOne(new LambdaUpdateWrapper<DeviceStatus>().eq(DeviceStatus::getDeviceId, deviceStatusDTO.getDeviceId()).eq(DeviceStatus::getDeleted, 0));DeviceStatus deviceStatus = new DeviceStatus();deviceStatus.setDeviceId(deviceStatusDTO.getDeviceId());deviceStatus.setDeviceName(deviceStatusDTO.getDeviceName());deviceStatus.setStatus(deviceStatusDTO.getStatus());deviceStatus.setLastUpdateTime(LocalDateTime.now());// 设置温度和湿度（仅当有值时）if (Objects.nonNull(deviceStatusDTO.getTemperature())) {deviceStatus.setTemperature(BigDecimal.valueOf(deviceStatusDTO.getTemperature()));}if (Objects.nonNull(deviceStatusDTO.getHumidity())) {deviceStatus.setHumidity(BigDecimal.valueOf(deviceStatusDTO.getHumidity()));}if (ObjectUtils.isEmpty(existingDevice)) {// 新增设备状态int rows = deviceStatusMapper.insert(deviceStatus);if (rows <= 0) {throw new RuntimeException("新增设备状态失败");}log.info("新增设备状态成功，deviceId: {}", deviceStatusDTO.getDeviceId());} else {// 更新设备状态deviceStatus.setId(existingDevice.getId());int rows = deviceStatusMapper.updateById(deviceStatus);if (rows <= 0) {throw new RuntimeException("更新设备状态失败");}log.info("更新设备状态成功，deviceId: {}", deviceStatusDTO.getDeviceId());}// 通过WebSocket推送设备状态到前端webSocketService.broadcastDeviceStatus(deviceStatusDTO);}/*** 处理订单支付数据** @param data 通用数据传输对象*/private void processOrderPaymentData(GenericDataDTO data) {// 这里实现订单支付数据的处理逻辑log.info("处理订单支付数据，dataId: {}", data.getDataId());// 解析订单数据并进行业务处理...// 推送处理结果到前端webSocketService.broadcast(data);}/*** 更新数据状态** @param data 通用数据传输对象* @param status 状态：0-新采集 1-已处理 2-异常*/private void updateDataStatus(GenericDataDTO data, int status) {LambdaUpdateWrapper<DataCollection> updateWrapper = new LambdaUpdateWrapper<>();updateWrapper.eq(DataCollection::getDataType, data.getDataType()).eq(DataCollection::getSource, data.getSource()).eq(DataCollection::getCreateTime, data.getTimestamp()).set(DataCollection::getStatus, status).set(DataCollection::getProcessTime, LocalDateTime.now());dataCollectionMapper.update(null, updateWrapper);log.info("更新数据状态成功，dataId: {}, status: {}", data.getDataId(), status);}
}

MyBatis-Plus Mapper 接口

定义数据访问层接口：

package com.realtime.mapper;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.realtime.entity.DataCollection;
import org.apache.ibatis.annotations.Mapper;/*** 数据采集记录Mapper** @author ken*/
@Mapper
public interface DataCollectionMapper extends BaseMapper<DataCollection> {
}

package com.realtime.mapper;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.realtime.entity.DeviceStatus;
import org.apache.ibatis.annotations.Mapper;/*** 设备状态Mapper** @author ken*/
@Mapper
public interface DeviceStatusMapper extends BaseMapper<DeviceStatus> {
}

WebSocket 实时推送实现

WebSocket 提供了浏览器和服务器之间的全双工通信，是实现实时数据推送的理想选择。我们将使用 Spring WebSocket 实现服务器向客户端的实时数据推送。

WebSocket 配置

首先配置 WebSocket 相关参数：

package com.realtime.config;import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
import com.realtime.handler.DataWebSocketHandler;
import com.realtime.interceptor.WebSocketHandshakeInterceptor;
import lombok.RequiredArgsConstructor;/*** WebSocket配置类** @author ken*/
@Configuration
@EnableWebSocket
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketConfigurer {private final DataWebSocketHandler dataWebSocketHandler;private final WebSocketHandshakeInterceptor webSocketHandshakeInterceptor;/*** 注册WebSocket处理器** @param registry WebSocket处理器注册表*/@Overridepublic void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {// 注册WebSocket处理器，映射路径为/ws/data，允许跨域访问registry.addHandler(dataWebSocketHandler, "/ws/data").addInterceptors(webSocketHandshakeInterceptor).setAllowedOrigins("*");// 为不支持WebSocket的浏览器提供SockJS备用方案registry.addHandler(dataWebSocketHandler, "/sockjs/data").addInterceptors(webSocketHandshakeInterceptor).withSockJS();}
}

WebSocket 握手拦截器

实现 WebSocket 握手拦截器，用于验证客户端连接：

package com.realtime.interceptor;import jakarta.servlet.http.HttpSession;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.http.server.ServletServerHttpRequest;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;import java.util.Map;/*** WebSocket握手拦截器** @author ken*/
@Slf4j
public class WebSocketHandshakeInterceptor implements HandshakeInterceptor {/*** 握手前处理** @param request  请求对象* @param response 响应对象* @param handler  WebSocket处理器* @param attributes 握手属性* @return 是否允许握手*/@Overridepublic boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,WebSocketHandler handler, Map<String, Object> attributes) {log.info("开始WebSocket握手...");// 从请求中获取HttpSessionif (request instanceof ServletServerHttpRequest servletRequest) {HttpSession session = servletRequest.getServletRequest().getSession(false);if (session != null) {// 可以从Session中获取用户信息等String userId = (String) session.getAttribute("userId");if (userId != null) {attributes.put("userId", userId);log.info("WebSocket握手成功，用户ID: {}", userId);return true;}}}// 这里可以添加自定义的认证逻辑// 简化示例，允许所有连接log.info("WebSocket握手成功，匿名用户");return true;}/*** 握手后处理** @param request  请求对象* @param response 响应对象* @param handler  WebSocket处理器* @param exception 异常*/@Overridepublic void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,WebSocketHandler handler, Exception exception) {if (exception != null) {log.error("WebSocket握手异常", exception);} else {log.info("WebSocket握手完成");}}
}

WebSocket 处理器

实现 WebSocket 消息处理器，负责管理连接和发送消息：

package com.realtime.handler;import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;/*** WebSocket数据处理器** @author ken*/
@Slf4j
@Component
public class DataWebSocketHandler extends TextWebSocketHandler {/*** 存储所有活跃的WebSocket会话*/private static final Set<WebSocketSession> SESSIONS = new CopyOnWriteArraySet<>();/*** 用户ID与会话的映射*/private static final Map<String, WebSocketSession> USER_SESSIONS = new ConcurrentHashMap<>();/*** 连接建立后调用** @param session WebSocket会话*/@Overridepublic void afterConnectionEstablished(WebSocketSession session) {// 将新连接添加到会话集合SESSIONS.add(session);log.info("WebSocket连接建立，会话ID: {}, 当前连接数: {}", session.getId(), SESSIONS.size());// 获取用户ID并存储映射关系String userId = (String) session.getAttributes().get("userId");if (userId != null) {USER_SESSIONS.put(userId, session);log.info("用户WebSocket连接建立，用户ID: {}, 会话ID: {}", userId, session.getId());}// 发送连接成功消息try {sendMessageToSession(session, "WebSocket连接成功");} catch (IOException e) {log.error("发送连接成功消息失败", e);}}/*** 处理收到的消息** @param session 会话* @param message 消息*/@Overrideprotected void handleTextMessage(WebSocketSession session, TextMessage message) {String payload = message.getPayload();log.info("收到WebSocket消息，会话ID: {}, 内容: {}", session.getId(), payload);// 可以在这里处理客户端发送的消息try {// 示例：回复收到消息的确认sendMessageToSession(session, "已收到消息: " + payload);} catch (IOException e) {log.error("回复消息失败", e);}}/*** 连接关闭后调用** @param session 会话* @param status  关闭状态*/@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus status) {// 从会话集合中移除关闭的连接SESSIONS.remove(session);log.info("WebSocket连接关闭，会话ID: {}, 状态: {}, 当前连接数: {}", session.getId(), status, SESSIONS.size());// 移除用户映射String userId = (String) session.getAttributes().get("userId");if (userId != null) {USER_SESSIONS.remove(userId);log.info("用户WebSocket连接关闭，用户ID: {}", userId);}}/*** 发生错误时调用** @param session 会话* @param exception 异常*/@Overridepublic void handleTransportError(WebSocketSession session, Throwable exception) {log.error("WebSocket传输错误，会话ID: {}", session.getId(), exception);}/*** 向指定会话发送消息** @param session 会话* @param message 消息内容* @throws IOException IO异常*/public void sendMessageToSession(WebSocketSession session, String message) throws IOException {if (session != null && session.isOpen()) {session.sendMessage(new TextMessage(message));}}/*** 向指定会话发送JSON消息** @param session 会话* @param data 数据对象* @throws IOException IO异常*/public void sendJsonMessageToSession(WebSocketSession session, Object data) throws IOException {if (session != null && session.isOpen()) {String json = JSON.toJSONString(data);session.sendMessage(new TextMessage(json));}}/*** 向所有连接的客户端广播消息** @param message 消息内容*/public void broadcast(String message) {for (WebSocketSession session : SESSIONS) {try {sendMessageToSession(session, message);} catch (IOException e) {log.error("广播消息失败，会话ID: {}", session.getId(), e);}}}/*** 向所有连接的客户端广播JSON消息** @param data 数据对象*/public void broadcast(Object data) {String json = JSON.toJSONString(data);for (WebSocketSession session : SESSIONS) {try {sendMessageToSession(session, json);} catch (IOException e) {log.error("广播JSON消息失败，会话ID: {}", session.getId(), e);}}}/*** 向指定用户发送消息** @param userId 用户ID* @param message 消息内容*/public void sendToUser(String userId, String message) {WebSocketSession session = USER_SESSIONS.get(userId);if (session != null) {try {sendMessageToSession(session, message);} catch (IOException e) {log.error("向用户发送消息失败，用户ID: {}", userId, e);}} else {log.warn("用户未连接，无法发送消息，用户ID: {}", userId);}}/*** 获取当前连接数** @return 连接数*/public int getConnectionCount() {return SESSIONS.size();}
}

WebSocket 服务封装

封装 WebSocket 服务，提供更友好的接口：

package com.realtime.service;import com.realtime.dto.DeviceStatusDTO;
import com.realtime.dto.GenericDataDTO;
import com.realtime.handler.DataWebSocketHandler;
import com.realtime.vo.WebSocketMessageVO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;/*** WebSocket服务类，提供消息推送接口** @author ken*/
@Slf4j
@Service
@RequiredArgsConstructor
public class WebSocketService {private final DataWebSocketHandler webSocketHandler;/*** 广播通用数据消息** @param data 通用数据传输对象*/public void broadcast(GenericDataDTO data) {if (data == null) {log.warn("广播的数据为空，不进行操作");return;}WebSocketMessageVO message = new WebSocketMessageVO();message.setType("data." + data.getDataType());message.setTimestamp(System.currentTimeMillis());message.setData(data.getContent());webSocketHandler.broadcast(message);log.info("已广播数据消息，dataId: {}, type: {}", data.getDataId(), message.getType());}/*** 广播设备状态消息** @param deviceStatus 设备状态数据*/public void broadcastDeviceStatus(DeviceStatusDTO deviceStatus) {if (deviceStatus == null) {log.warn("广播的设备状态为空，不进行操作");return;}WebSocketMessageVO message = new WebSocketMessageVO();message.setType("device.status");message.setTimestamp(System.currentTimeMillis());message.setData(deviceStatus);webSocketHandler.broadcast(message);log.info("已广播设备状态消息，deviceId: {}", deviceStatus.getDeviceId());}/*** 向指定用户发送消息** @param userId 用户ID* @param message 消息内容*/public void sendToUser(String userId, String message) {if (!org.springframework.util.StringUtils.hasText(userId)) {log.warn("用户ID为空，不发送消息");return;}webSocketHandler.sendToUser(userId, message);log.info("已向用户发送消息，userId: {}", userId);}/*** 获取当前WebSocket连接数** @return 连接数*/public int getConnectionCount() {return webSocketHandler.getConnectionCount();}
}

WebSocket 消息 VO

定义 WebSocket 消息的封装类：

package com.realtime.vo;import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;/*** WebSocket消息VO** @author ken*/
@Data
@Schema(description = "WebSocket消息VO")
public class WebSocketMessageVO {/*** 消息类型*/@Schema(description = "消息类型", example = "device.status, order.payment")private String type;/*** 消息时间戳（毫秒）*/@Schema(description = "消息时间戳（毫秒）")private Long timestamp;/*** 消息数据*/@Schema(description = "消息数据")private Object data;
}

前端看板实现（Vue 3 + ECharts）

前端看板负责接收 WebSocket 推送的数据并进行可视化展示。我们使用 Vue 3 和 ECharts 实现一个实时设备监控看板。

前端项目初始化

  1. 创建 Vue 项目：
npm create vue@latest realtime-dashboard-frontend
cd realtime-dashboard-frontend
npm install
npm install echarts sockjs-client stompjs
  1. 安装必要的依赖：
npm install echarts sockjs-client @stomp/stompjs

WebSocket 服务封装

创建 WebSocket 服务，处理与后端的连接和消息接收：

javascript

// src/services/webSocketService.js
import SockJS from 'sockjs-client';
import { Client } from '@stomp/stompjs';class WebSocketService {constructor() {this.client = null;this.connected = false;this.subscribers = new Map(); // 存储不同消息类型的订阅者}/*** 连接WebSocket服务器* @param {string} url - WebSocket连接地址* @returns {Promise} - 连接结果*/connect(url) {return new Promise((resolve, reject) => {// 如果已经连接，则直接返回成功if (this.connected) {resolve(true);return;}// 创建SockJS实例const socket = new SockJS(url);// 创建STOMP客户端this.client = new Client({webSocketFactory: () => socket,reconnectDelay: 5000, // 重连延迟（毫秒）heartbeatIncoming: 4000,heartbeatOutgoing: 4000,});// 连接成功回调this.client.onConnect = (frame) => {console.log('WebSocket连接成功:', frame);this.connected = true;resolve(true);};// 连接错误回调this.client.onStompError = (frame) => {console.error('WebSocket连接错误:', frame.headers['message']);reject(new Error(frame.headers['message'] || 'WebSocket连接错误'));};// 连接关闭回调this.client.onDisconnect = () => {console.log('WebSocket连接已关闭');this.connected = false;};// 消息接收回调this.client.onMessage = (message) => {try {const data = JSON.parse(message.body);this.handleReceivedMessage(data);} catch (error) {console.error('解析WebSocket消息失败:', error, message.body);}};// 启动连接this.client.activate();});}/*** 断开WebSocket连接*/disconnect() {if (this.client && this.connected) {this.client.deactivate();this.connected = false;console.log('WebSocket已断开连接');}}/*** 处理接收到的消息* @param {Object} message - 消息对象*/handleReceivedMessage(message) {if (!message || !message.type) {console.warn('无效的WebSocket消息:', message);return;}const { type, data } = message;console.log(`收到消息，类型: ${type}`, data);// 通知该类型消息的所有订阅者if (this.subscribers.has(type)) {const callbacks = this.subscribers.get(type);callbacks.forEach(callback => {try {callback(data);} catch (error) {console.error(`处理${type}类型消息失败:`, error);}});} else {console.log(`没有订阅者订阅${type}类型的消息`);}}/*** 订阅指定类型的消息* @param {string} type - 消息类型* @param {Function} callback - 消息处理回调* @returns {Function} - 取消订阅的函数*/subscribe(type, callback) {if (!type || typeof callback !== 'function') {console.error('无效的订阅参数');return () => {};}if (!this.subscribers.has(type)) {this.subscribers.set(type, new Set());}const callbacks = this.subscribers.get(type);callbacks.add(callback);// 返回取消订阅的函数return () => {callbacks.delete(callback);if (callbacks.size === 0) {this.subscribers.delete(type);}};}/*** 发送消息到服务器* @param {string} destination - 目的地* @param {Object} message - 消息内容*/send(destination, message) {if (!this.connected) {console.error('WebSocket未连接，无法发送消息');return;}try {this.client.publish({destination,body: JSON.stringify(message)});console.log(`发送消息到${destination}:`, message);} catch (error) {console.error('发送WebSocket消息失败:', error);}}/*** 检查连接状态* @returns {boolean} - 是否连接*/isConnected() {return this.connected;}
}// 导出单例实例
export default new WebSocketService();

设备监控看板组件

实现一个实时设备监控看板，展示设备状态和实时数据：

<!-- src/views/DeviceDashboard.vue -->
<template><div class="dashboard-container"><div class="dashboard-header"><h1>实时设备监控看板</h1><div class="status-info"><span :class="{'connected': isConnected, 'disconnected': !isConnected}">{{ isConnected ? '已连接' : '未连接' }}</span><span>连接数: {{ connectionCount }}</span><span>最后更新: {{ lastUpdateTime || '无数据' }}</span></div></div><div class="dashboard-grid"><!-- 设备状态概览 --><div class="dashboard-card"><h2>设备状态概览</h2><div class="status-summary"><div class="status-item online"><span class="status-label">在线设备</span><span class="status-value">{{ onlineCount }}</span></div><div class="status-item offline"><span class="status-label">离线设备</span><span class="status-value">{{ offlineCount }}</span></div><div class="status-item error"><span class="status-label">异常设备</span><span class="status-value">{{ errorCount }}</span></div><div class="status-item total"><span class="status-label">总设备数</span><span class="status-value">{{ totalCount }}</span></div></div></div><!-- 温度趋势图 --><div class="dashboard-card"><h2>设备温度趋势</h2><div class="chart-container"><div ref="temperatureChart" class="chart"></div></div></div><!-- 湿度趋势图 --><div class="dashboard-card"><h2>设备湿度趋势</h2><div class="chart-container"><div ref="humidityChart" class="chart"></div></div></div><!-- 设备列表 --><div class="dashboard-card device-list-card"><h2>设备列表</h2><div class="device-table"><table><thead><tr><th>设备ID</th><th>设备名称</th><th>状态</th><th>温度 (°C)</th><th>湿度 (%)</th><th>信号强度</th><th>最后更新</th></tr></thead><tbody><tr v-for="device in devices" :key="device.deviceId" :class="getStatusClass(device.status)"><td>{{ device.deviceId }}</td><td>{{ device.deviceName }}</td><td>{{ getStatusText(device.status) }}</td><td>{{ device.temperature !== null ? device.temperature.toFixed(1) : '-' }}</td><td>{{ device.humidity !== null ? device.humidity.toFixed(1) : '-' }}</td><td>{{ device.signalStrength || '-' }}</td><td>{{ formatTime(device.collectTime) }}</td></tr></tbody></table></div></div></div></div>
</template><script setup>
import { ref, onMounted, onUnmounted, reactive, toRefs, watch } from 'vue';
import * as echarts from 'echarts';
import webSocketService from '../services/webSocketService';
import { format } from 'date-fns';// 状态数据
const state = reactive({devices: [], // 设备列表isConnected: false, // WebSocket连接状态connectionCount: 0, // 连接数lastUpdateTime: null, // 最后更新时间temperatureData: new Map(), // 温度数据humidityData: new Map() // 湿度数据
});// 解构响应式数据
const { devices, isConnected, connectionCount, lastUpdateTime, temperatureData, humidityData } = toRefs(state);// 图表实例
const temperatureChart = ref(null);
const humidityChart = ref(null);
let tempChartInstance = null;
let humiChartInstance = null;// 计算属性 - 设备状态统计
const onlineCount = ref(0);
const offlineCount = ref(0);
const errorCount = ref(0);
const totalCount = ref(0);// 监听设备列表变化，更新统计数据
watch(devices, (newDevices) => {onlineCount.value = newDevices.filter(d => d.status === 1).length;offlineCount.value = newDevices.filter(d => d.status === 0).length;errorCount.value = newDevices.filter(d => d.status === 2).length;totalCount.value = newDevices.length;
}, { deep: true });/*** 初始化WebSocket连接*/
const initWebSocket = () => {// 连接WebSocket服务器const wsUrl = 'http://localhost:8080/api/sockjs/data';webSocketService.connect(wsUrl).then(() => {state.isConnected = true;console.log('WebSocket连接成功');// 订阅设备状态消息const unsubscribe = webSocketService.subscribe('device.status', handleDeviceStatusMessage);// 组件卸载时取消订阅onUnmounted(() => {unsubscribe();});}).catch(error => {console.error('WebSocket连接失败:', error);state.isConnected = false;});// 监听连接状态变化const checkConnectionStatus = setInterval(() => {state.isConnected = webSocketService.isConnected();}, 1000);onUnmounted(() => {clearInterval(checkConnectionStatus);});
};/*** 处理设备状态消息* @param {Object} data - 设备状态数据*/
const handleDeviceStatusMessage = (data) => {if (!data || !data.deviceId) {console.warn('无效的设备状态数据:', data);return;}// 更新最后更新时间state.lastUpdateTime = format(new Date(), 'yyyy-MM-dd HH:mm:ss');// 查找设备在列表中的位置const index = state.devices.findIndex(d => d.deviceId === data.deviceId);if (index > -1) {// 更新现有设备state.devices[index] = { ...state.devices[index], ...data };} else {// 添加新设备state.devices.push(data);}// 更新图表数据updateChartData(data);
};/*** 更新图表数据* @param {Object} deviceData - 设备数据*/
const updateChartData = (deviceData) => {if (deviceData.status !== 1) {// 设备不在线，不更新数据return;}const time = format(new Date(deviceData.collectTime), 'HH:mm:ss');// 更新温度数据if (deviceData.temperature !== null) {if (!temperatureData.value.has(deviceData.deviceId)) {temperatureData.value.set(deviceData.deviceId, {name: deviceData.deviceName,data: []});}const tempData = temperatureData.value.get(deviceData.deviceId);tempData.data.push([time, deviceData.temperature]);// 保持数据点不超过30个if (tempData.data.length > 30) {tempData.data.shift();}}// 更新湿度数据if (deviceData.humidity !== null) {if (!humidityData.value.has(deviceData.deviceId)) {humidityData.value.set(deviceData.deviceId, {name: deviceData.deviceName,data: []});}const humiData = humidityData.value.get(deviceData.deviceId);humiData.data.push([time, deviceData.humidity]);// 保持数据点不超过30个if (humiData.data.length > 30) {humiData.data.shift();}}// 刷新图表refreshCharts();
};/*** 初始化图表*/
const initCharts = () => {// 初始化温度图表tempChartInstance = echarts.init(temperatureChart.value);tempChartInstance.setOption({title: { text: '实时温度监测' },tooltip: {trigger: 'axis',axisPointer: { type: 'cross' }},legend: { data: [] },xAxis: { type: 'category', boundaryGap: false },yAxis: { type: 'value', name: '温度 (°C)' },series: []});// 初始化湿度图表humiChartInstance = echarts.init(humidityChart.value);humiChartInstance.setOption({title: { text: '实时湿度监测' },tooltip: {trigger: 'axis',axisPointer: { type: 'cross' }},legend: { data: [] },xAxis: { type: 'category', boundaryGap: false },yAxis: { type: 'value', name: '湿度 (%)' },series: []});// 监听窗口大小变化，调整图表大小const handleResize = () => {tempChartInstance.resize();humiChartInstance.resize();};window.addEventListener('resize', handleResize);onUnmounted(() => {window.removeEventListener('resize', handleResize);});
};/*** 刷新图表数据*/
const refreshCharts = () => {if (!tempChartInstance || !humiChartInstance) {return;}// 更新温度图表const tempSeries = [];const tempLegend = [];temperatureData.value.forEach((value, key) => {tempLegend.push(value.name);tempSeries.push({name: value.name,type: 'line',data: value.data,smooth: true,symbol: 'none'});});tempChartInstance.setOption({legend: { data: tempLegend },series: tempSeries});// 更新湿度图表const humiSeries = [];const humiLegend = [];humidityData.value.forEach((value, key) => {humiLegend.push(value.name);humiSeries.push({name: value.name,type: 'line',data: value.data,smooth: true,symbol: 'none'});});humiChartInstance.setOption({legend: { data: humiLegend },series: humiSeries});
};/*** 获取状态文本* @param {number} status - 状态码* @returns {string} 状态文本*/
const getStatusText = (status) => {switch (status) {case 0: return '离线';case 1: return '在线';case 2: return '异常';default: return '未知';}
};/*** 获取状态样式类名* @param {number} status - 状态码* @returns {string} 样式类名*/
const getStatusClass = (status) => {switch (status) {case 0: return 'offline-row';case 1: return 'online-row';case 2: return 'error-row';default: return '';}
};/*** 格式化时间* @param {string} timeStr - 时间字符串* @returns {string} 格式化后的时间*/
const formatTime = (timeStr) => {if (!timeStr) return '-';return format(new Date(timeStr), 'yyyy-MM-dd HH:mm:ss');
};// 组件挂载时初始化
onMounted(() => {initWebSocket();initCharts();
});// 组件卸载时清理
onUnmounted(() => {webSocketService.disconnect();if (tempChartInstance) {tempChartInstance.dispose();}if (humiChartInstance) {humiChartInstance.dispose();}
});
</script><style scoped>
.dashboard-container {padding: 20px;background-color: #f5f5f5;min-height: 100vh;
}.dashboard-header {display: flex;justify-content: space-between;align-items: center;margin-bottom: 20px;
}.status-info {display: flex;gap: 20px;font-size: 14px;
}.status-info .connected {color: #4caf50;
}.status-info .disconnected {color: #f44336;
}.dashboard-grid {display: grid;grid-template-columns: repeat(2, 1fr);gap: 20px;
}.dashboard-card {background-color: white;border-radius: 8px;box-shadow: 0 2px 8px rgba(0, 0, 0, 0.1);padding: 20px;
}.dashboard-card h2 {margin-top: 0;font-size: 18px;color: #333;border-bottom: 1px solid #eee;padding-bottom: 10px;margin-bottom: 15px;
}.status-summary {display: grid;grid-template-columns: repeat(4, 1fr);gap: 10px;text-align: center;
}.status-item {padding: 15px 10px;border-radius: 6px;color: white;
}.status-item .status-label {display: block;font-size: 14px;margin-bottom: 5px;
}.status-item .status-value {font-size: 24px;font-weight: bold;
}.status-item.online {background-color: #4caf50;
}.status-item.offline {background-color: #9e9e9e;
}.status-item.error {background-color: #f44336;
}.status-item.total {background-color: #2196f3;
}.chart-container {width: 100%;height: 300px;
}.chart {width: 100%;height: 100%;
}.device-list-card {grid-column: span 2;
}.device-table {overflow-x: auto;
}.device-table table {width: 100%;border-collapse: collapse;min-width: 800px;
}.device-table th,
.device-table td {padding: 12px 15px;text-align: left;border-bottom: 1px solid #eee;
}.device-table th {background-color: #f9f9f9;font-weight: bold;
}.online-row {background-color: rgba(76, 175, 80, 0.05);
}.offline-row {background-color: rgba(158, 158, 158, 0.05);
}.error-row {background-color: rgba(244, 67, 54, 0.05);
}@media (max-width: 1200px) {.dashboard-grid {grid-template-columns: 1fr;}.device-list-card {grid-column: span 1;}
}
</style>

路由配置

配置前端路由：

// src/router/index.js
import { createRouter, createWebHistory } from 'vue-router';
import DeviceDashboard from '../views/DeviceDashboard.vue';const router = createRouter({history: createWebHistory(import.meta.env.BASE_URL),routes: [{path: '/',name: 'dashboard',component: DeviceDashboard}]
});export default router;

系统测试与优化

功能测试

  1. RabbitMQ 消息测试：

    • 启动 RabbitMQ 管理界面，观察交换机和队列是否正确创建
    • 启动应用程序，查看设备数据采集器是否正常发送消息
    • 在 RabbitMQ 管理界面中查看消息是否被正确路由到相应队列

  2. WebSocket 连接测试：

    • 启动前端应用，打开浏览器控制台，检查 WebSocket 是否连接成功
    • 观察是否能接收到后端推送的设备状态数据
    • 测试多客户端连接，确认广播功能正常

  3. 数据处理测试：

    • 查看数据库，确认数据是否被正确存储
    • 检查异常数据是否被正确路由到错误队列
    • 验证数据处理逻辑是否符合预期

性能优化

  1. RabbitMQ 优化：

    • 根据业务需求调整消费者并发数
    • 设置合理的 prefetch 值，避免消费者过载
    • 为队列设置合理的 TTL（生存时间）和死信策略

  2. WebSocket 优化：

    • 实现消息压缩，减少数据传输量
    • 对频繁更新的数据进行节流处理
    • 实现客户端心跳检测，及时清理无效连接

  3. 数据库优化：

    • 为常用查询字段建立索引
    • 对历史数据进行分区存储
    • 实现数据归档策略，避免表过大影响性能

  4. 前端优化：

    • 对图表数据进行采样处理，避免数据点过多影响渲染性能
    • 使用虚拟滚动优化长列表展示
    • 实现数据缓存，减少重复渲染

总结与扩展

本文详细介绍了如何利用 RabbitMQ 和 WebSocket 构建实时数据看板系统，从架构设计到代码实现，涵盖了数据采集、消息传递、数据处理和前端展示的完整流程。通过这个系统，你可以实时采集多源数据，经过处理后推送到前端看板，实现数据的可视化监控。

http://www.dtcms.com/a/457358.html

相关文章：

  • Linux（含嵌入式设备如泰山派）VNC 完整配置指南：含开机自启动（适配 Ubuntu/Debian 系）
  • 网站营销活动泰安市高新区建设局网站
  • 玳瑁的嵌入式日记 --------API总结
  • [xboard] 26 kernel启动流程之initrd、initramfs、ramdisk核心异同
  • 鸿蒙实现滴滴出行项目之侧边抽屉栏以及权限以及搜索定位功能
  • 从OpenAI发布会看AI未来：中国就业市场的重构与突围
  • 乔拓云网站建设wps怎么做网站
  • TensorFlow2 Python深度学习 - TensorFlow2框架入门 - 立即执行模式（Eager Execution）
  • 监控系统2 - framebuffer
  • 大型网站响应式wordpress3.8.3
  • Django `models.Field` 所有常见配置参数的完整清单与说明表
  • 音乐播放器-单html文件
  • 18+网站推广平面设计师素材网站
  • 告别传统SEO！拥抱下一代流量密码：生成式引擎优化（GEO）实战指南
  • 不抽佣美团核销API接口，同时支持美团抖音核销,快速接入
  • 优化网站关键词的技巧建设银行忘记密码网站
  • 网站建设属于服务还是货物为什么asp.net做的网站上传后不显示照片
  • TT直流减速电机（STM32）
  • MQTT协议三个核心函数的详解
  • 马拉松网站建设方案汽车网站管理系统
  • 【企业架构】TOGAF架构标准规范-技术架构
  • 【Proteus仿真】甲单片机(AT89C51)控制乙单片机(AT89C51)
  • 南阳网站推广优化公司discuz和wordpress区别
  • XSLT `＜value-of＞` 标签详解
  • 做外贸网站注册什么邮箱懒人模板网站
  • 案例-使用Postman向百度首页发送带时间戳的GET请求
  • leetcode排序链表
  • 做经营性的网站备案条件汉字叔叔花了多少钱做网站
  • 效果好企业营销型网站建设手机html5网站源码
  • 浏览器事件循环与内存管理可视化