实现一个基于 Spring Boot 和 Vue.js 的实时消息推送系统

在现代互联网应用中，实时消息推送已经成为一个非常重要的功能。不论是即时通讯、通知系统，还是其他需要实时互动的应用场景，消息的实时性直接影响到用户的体验和应用的效率。在这篇文章中，我将详细介绍如何使用 Spring Boot 和 Vue.js 创建一个实时消息推送系统，并确保每个用户只能接收属于自己的消息。这个系统不仅功能强大，而且实现起来并不复杂。

项目概述

我们的系统主要包括以下几个功能：

实时消息推送。
消息的持久化存储（使用 MySQL 和 Redis）。
用户的历史消息查看。

技术栈

为了实现上述功能，我们需要使用以下技术：

后端：Spring Boot、WebSocket、MySQL、Redis
前端：Vue.js、SockJS、STOMP

关键概念介绍

在正式开始实现之前，了解一些关键概念是非常重要的：

1. WebSocket

WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。它允许服务器和客户端之间实时地交换数据，从而大大提升了数据传输的效率。与传统的 HTTP 不同，WebSocket 是一个持久化的连接，可以在客户端和服务器之间建立一个持续的双向通信通道。这样，无论是服务器向客户端推送消息，还是客户端向服务器发送数据，都能做到实时、低延迟。

2. STOMP

STOMP（Simple Text Oriented Messaging Protocol）是一个简单的、基于文本的消息传递协议。STOMP 协议本身不依赖于任何特定的传输协议，因此可以在多种传输协议上使用，如 TCP、WebSocket 等。STOMP 的引入简化了我们在 WebSocket 上处理消息订阅和发布的工作，它使得消息的管理变得更加直观和简单。

3. SockJS

SockJS 是一个用于 WebSockets 的 JavaScript 库。它提供了一个 WebSocket-like 的接口，并且在 WebSocket 不可用时能够回退到其他传输方式（如 AJAX 长轮询）。SockJS 确保了在各种网络条件和浏览器环境下都能正常工作，为我们的实时消息推送系统提供了坚实的基础。

项目实现步骤

步骤一：后端实现

1. WebSocket 配置

首先，我们需要配置 WebSocket，以便实现消息的实时推送。在 Spring Boot 中，我们可以通过 WebSocketConfig 类来进行配置。这部分代码的核心在于启用 STOMP 消息代理，并注册 WebSocket 端点。

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic", "/queue");
        config.setApplicationDestinationPrefixes("/app");
        config.setUserDestinationPrefix("/user");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").setAllowedOrigins("*").withSockJS();
    }
}

2. 消息实体类

为了保存和管理消息，我们需要创建一个实体类 Message。这个实体类将用于存储消息的基本信息，例如发送者、接收者、内容和时间戳。

@Entity
@Table(name = "messages")
public class Message {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String fromUser;
    private String toUser;
    private String topic;
    private String content;

    @Column(name = "timestamp", nullable = false, updatable = false)
    private LocalDateTime timestamp;

    // Getters and setters
}

3. 消息服务

消息服务是实现消息发送和存储的核心部分。我们需要确保每次发送消息时，消息不仅会被推送到接收者，还会被保存到数据库中，以便后续查看历史消息。

@Service
public class MessageService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    private MessageRepository messageRepository;

    public void sendMessage(String fromUser, String toUser, String topic, String content) {
        // 保存消息到Redis
        redisTemplate.opsForList().leftPush(topic, content);

        // 保存消息到MySQL
        Message message = new Message();
        message.setFromUser(fromUser);
        message.setToUser(toUser);
        message.setTopic(topic);
        message.setContent(content);
        message.setTimestamp(LocalDateTime.now());
        messageRepository.save(message);

        // 推送消息到接收者客户端
        messagingTemplate.convertAndSend("/queue/messages", message);
    }

    // 从Redis获取消息
    public List<Object> getMessagesFromRedis(String topic) {
        return redisTemplate.opsForList().range(topic, 0, -1);
    }

    // 从MySQL获取历史消息
    public List<Message> getMessagesFromDatabase(String topic) {
        return messageRepository.findByTopicOrderByTimestampDesc(topic);
    }
}

4. 消息控制器

消息控制器用于处理客户端的请求，包括发送消息和获取历史消息。

@RestController
@RequestMapping("/messages")
public class MessageController {

    @Autowired
    private MessageService messageService;

    @PostMapping("/send")
    public void sendMessage(@RequestParam String fromUser, @RequestParam String toUser, @RequestParam String topic, @RequestParam String content) {
        messageService.sendMessage(fromUser, toUser, topic, content);
    }

    @GetMapping("/history")
    public List<Message> getHistoryMessages(@RequestParam String topic) {
        return messageService.getMessagesFromDatabase(topic);
    }
}

步骤二：前端实现

前端部分需要实现 WebSocket 的连接和消息的发送与接收。

1. WebSocket 服务

WebSocket 服务用于管理 WebSocket 的连接和消息的发送。

// src/services/websocket-service.js
import SockJS from 'sockjs-client';
import Stomp from 'stompjs';

let stompClient = null;

export function connect(userId, onMessageReceived) {
    const socket = new SockJS('/ws');
    stompClient = Stomp.over(socket);
    stompClient.connect({}, frame => {
        console.log('Connected: ' + frame);
        stompClient.subscribe(`/queue/messages`, message => {
            onMessageReceived(JSON.parse(message.body));
        });
    });
}

export function sendMessage(fromUser, toUser, topic, content) {
    const message = {
        fromUser,
        toUser,
        topic,
        content,
        timestamp: new Date().toISOString()
    };
    stompClient.send('/app/send', {}, JSON.stringify(message));
}

2. API 服务

API 服务用于与后端进行 HTTP 请求。

// src/services/api-service.js
import axios from 'axios';

export function fetchHistoryMessages(topic) {
    return axios.get('/messages/history', { params: { topic } });
}

export function sendMessage(fromUser, toUser, topic, content) {
    return axios.post('/messages/send', null, { params: { fromUser, toUser, topic, content } });
}

3. Vue 组件

Vue 组件用于展示消息并与用户交互。

<template>
    <div>
        <div>
            <h2>历史消息</h2>
            <div v-for="message in historyMessages" :key="message.id">
                <strong>{{ message.timestamp }} ({{ message.fromUser }} -> {{ message.toUser }}):</strong> {{ message.content }}
            </div>
        </div>
        <div>
            <h2>实时消息</h2>
            <div v-for="message in messages" :key="message.timestamp">
                <strong>{{ message.timestamp }} ({{ message.fromUser }} -> {{ message.toUser }}):</strong> {{ message.content }}
            </div>
            <input v-model="newMessage" @keyup.enter="sendMessage">
        </div>
    </div>
</template>

<script>
import { connect, sendMessage } from '@/services/websocket-service';
import { fetchHistoryMessages } from '@/services/api-service';

export default {
    data() {
        return {
            newMessage: '',
            messages: [],
            historyMessages: [],
            fromUser: 'user1', // should be dynamic based on your app logic
            toUser: 'user2',   // should be dynamic based on your app logic
            topic: 'topic1'    // should be dynamic based on your app logic
        };
    },
    created() {
        connect(this.fromUser, message => {
            this.messages.push(message);
        });

        // 获取历史消息
        fetchHistoryMessages(this.topic).then(response => {
            this.historyMessages = response.data;
        });
    },
    methods: {
        sendMessage() {
            sendMessage(this.fromUser, this.toUser, this.topic, this.newMessage).then(() => {
                this.newMessage = '';
            });
        }
    }
};
</script>