spring+rabbitmq+stomp搭建websocket消息推送(非springbo…

Spring + RabbitMQ + Stomp 搭建 WebSocket 消息推送（非 Spring Boot 版本）

WebSocket 是一项在 Web 开发中非常重要的技术，它允许服务器和客户端之间实时、双向通信。在实际开发过程中，我们通常需要使用一些消息队列来实现后台消息推送系统，而 RabbitMQ 是一个非常优秀的消息队列实现。本文将介绍如何使用 Spring 和 RabbitMQ，以及使用 Stomp 协议来搭建 WebSocket 消息推送系统。

项目搭建

本项目使用的技术栈是 Spring + RabbitMQ + Stomp。为了测试这个应用，我们需要前端页面发送一个消息，然后后端通过 RabbitMQ 将消息传递给具有 WebSocket 连接的客户端。下面开始搭建这个应用。

创建项目

创建一个 Maven 项目，并在 pom.xml 文件中加入以下依赖：

<dependencies>
    <!-- Spring 相关依赖 -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-core</artifactId>
        <version>${spring-framework.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-messaging</artifactId>
        <version>${spring-framework.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-websocket</artifactId>
        <version>${spring-framework.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>${spring-amqp.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-amqp-stomp</artifactId>
        <version>${spring-amqp.version}</version>
    </dependency>

    <!-- RabbitMQ 相关依赖 -->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>${rabbitmq.version}</version>
    </dependency>

    <!-- Stomp JavaScript 客户端 -->
    <dependency>
        <groupId>org.webjars</groupId>
        <artifactId>stomp-websocket</artifactId>
        <version>${stomp-websocket.version}</version>
    </dependency>
</dependencies>

其中 ${spring-framework.version}、${spring-amqp.version}、${rabbitmq.version}、${stomp-websocket.version} 定义了各种依赖的版本号，开发者可以使用自己项目需要的版本号来替换。

创建 RabbitMQ Exchange 和 Queue

在 RabbitMQ 中，Exchange 和 Queue 存储了实际传递的消息。在本应用中，我们需要创建一个 Exchange 和一个 Queue 用来存储消息。下面是代码：

@Configuration
public class RabbitMqConfig {

    public static final String EXCHANGE_NAME = "websocket-exchange";

    public static final String QUEUE_NAME = "websocket-queue";

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, false);
    }

    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("#");
    }
}

配置 RabbitMQ

在 application.properties 文件中加入以下配置：

# RabbitMQ 配置
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

创建 WebSocket Endpoint

在 Spring 中，我们需要创建一个 WebSocket Endpoint，通过它来处理客户端的请求。这个 Endpoint 需要处理 connect 和 disconnect Event，还需要处理来自客户端的消息。以下是相应的代码：

@Component
public class WebSocketEndpoint extends TextWebSocketHandler {

    private final RabbitTemplate rabbitTemplate;

    @Autowired
    public WebSocketEndpoint(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        System.out.println("WebSocket 连接已建立");
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        System.out.println("WebSocket 连接已关闭");
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        System.out.println("From client: " + payload);
    }
}

创建 RabbitMQ Consumer

在 RabbitMQ 中，我们需要为 Queue 建立一个 Consumer，以便在有新消息到达时可以处理它。以下是相应的代码：

@Component
public class RabbitMqConsumer implements MessageListener {

    private final SimpMessagingTemplate messagingTemplate;

    @Autowired
    public RabbitMqConsumer(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @Override
    public void onMessage(Message message) {
        String payload = new String(message.getBody());
        System.out.println("From RabbitMQ: " + payload);

        messagingTemplate.convertAndSend("/topic/greetings", payload);
    }
}

配置 WebSocket

在 WebSocketConfig.java 中定义了两个 Bean 对象：WebSocketHandlerMapping 和 MessageBroker. WebSocketHandlerMapping 用来处理 WebSocket 请求，MessageBroker 用来将消息传递到 Broker 中。以下是相应的代码：

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Autowired
    private WebSocketEndpoint webSocketEndpoint;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(webSocketEndpoint, "/websocket");
    }

    @Bean
    public WebSocketMessageBrokerConfigurer configureMessageBroker() {
        return new WebSocketMessageBrokerConfigurer() {

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

            @Override
            public void configureMessageBroker(MessageBrokerRegistry registry) {
                registry.setApplicationDestinationPrefixes("/app");
                registry.enableStompBrokerRelay("/topic")
                        .setRelayHost("localhost")
                        .setRelayPort(61613)
                        .setClientLogin("guest")
                        .setClientPasscode("guest")
                        .setSystemLogin("guest")
                        .setSystemPasscode("guest");
            }
        };
    }
}

前端页面

下面是前端页面的代码，它使用 Stomp.js 客户端来和后端交互。

<!DOCTYPE html>
<html>
<head>
    <title>WebSocket</title>
    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
    <script src="/webjars/stomp-websocket/stomp.min.js"></script>

    <script>
        var stompClient = null;

        function connect() {
            var socket = new SockJS('/stomp');
            stompClient = Stomp.over(socket);
            stompClient.connect({}, function(frame) {
                console.log('Connected: ' + frame);

                stompClient.subscribe('/topic/greetings', function (greeting) {
                    $('#greetings').append("<tr><td>" + JSON.parse(greeting.body).content + "</td></tr>");
                });
            });
        }

        function disconnect() {
            if (stompClient != null) {
                stompClient.disconnect();
            }
            console.log("Disconnected.");
        }

        function sendName() {
            stompClient.send("/app/hello", {}, JSON.stringify({'name': $("#name").val()}));
        }
    </script>
</head>

<body>
    <div>
        <input type="text" id="name" placeholder="Your name here">
        <button onclick="sendName()">Send</button>
    </div>
    <table>
        <thead>
            <tr>
                <th>Greetings</th>
            </tr>
        </thead>
        <tbody id="greetings">
        </tbody>
    </table>
    <div>
        <button onclick="connect()">Connect</button>
        <button onclick="disconnect()">Disconnect</button>
    </div>
</body>

</html>

测试

为了测试本应用，我们需要启动 RabbitMQ 服务器，并启动本应用。在 WebSocketEndpoint 中打开注释 rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_NAME, "test message");，然后在浏览器中访问 http://localhost:8080，点击“Connect”按钮，此时应该看到 WebSocket 连接已建立的输出信息。现在，再在 WebSocketEndpoint 中打开注释之前的代码段，然后点击“Send”按钮，再打开控制台，应该看到来自 RabbitMQ 的消息已经发送到了客户端。

结论

本文介绍了如何使用 Spring 和 RabbitMQ，以及使用 Stomp 协议来搭建 WebSocket 消息推送系统。通过这个应用，我们了解了如何使用 WebSocket、RabbitMQ 和 Stomp 来搭建一个实时应用。