消息流

ChatGPT-4o-mini 中英对照 Flow of Messages

一旦 STOMP 端点被暴露，Spring 应用程序就成为连接客户端的 STOMP 代理。本节描述了服务器端消息的流动。

spring-messaging 模块包含了对消息应用程序的基础支持，这些应用程序起源于 Spring Integration，后来被提取并纳入 Spring Framework，以便在许多 Spring 项目 和应用场景中更广泛地使用。以下列表简要描述了一些可用的消息抽象：

• Message: 消息的简单表示，包括头信息和负载。

• MessageHandler: 处理消息的契约。

• MessageChannel: 发送消息的契约，使生产者和消费者之间实现松耦合。

• SubscribableChannel: 带有 MessageHandler 订阅者的 MessageChannel。

• ExecutorSubscribableChannel: 使用 Executor 传递消息的 SubscribableChannel。

Java 配置（即 @EnableWebSocketMessageBroker）和 XML 命名空间配置（即 <websocket:message-broker>）都使用前面的组件来组装消息工作流。下图显示了启用简单内置消息代理时使用的组件：

上面的图显示了三个消息通道：

• clientInboundChannel: 用于传递从 WebSocket 客户端接收到的消息。

• clientOutboundChannel: 用于将服务器消息发送到 WebSocket 客户端。

• brokerChannel: 用于从服务器端应用程序代码向消息代理发送消息。

下图显示了在配置外部代理（如 RabbitMQ）以管理订阅和广播消息时使用的组件：

两个前面图表之间的主要区别在于使用“代理中继”通过 TCP 将消息传递到外部 STOMP 代理，以及将消息从代理传递到已订阅的客户端。

当从 WebSocket 连接接收到消息时，它们会被解码为 STOMP 帧，转化为 Spring Message 表示，并发送到 clientInboundChannel 以进行进一步处理。例如，目标头以 /app 开头的 STOMP 消息可能会被路由到注解控制器中的 @MessageMapping 方法，而 /topic 和 /queue 消息可能会直接路由到消息代理。

一个注解的 @Controller 处理来自客户端的 STOMP 消息，可以通过 brokerChannel 向消息代理发送消息，然后代理通过 clientOutboundChannel 向匹配的订阅者广播消息。相同的控制器也可以在响应 HTTP 请求时执行相同的操作，因此客户端可以执行 HTTP POST，然后 @PostMapping 方法可以向消息代理发送消息，以广播给已订阅的客户端。

我们可以通过一个简单的例子来追踪流程。考虑以下例子，它设置了一个服务器：

Java

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfiguration implements WebSocketMessageBrokerConfigurer {

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

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.setApplicationDestinationPrefixes("/app");
        registry.enableSimpleBroker("/topic");
    }
}

java

Kotlin

@Configuration
@EnableWebSocketMessageBroker
class WebSocketConfiguration : WebSocketMessageBrokerConfigurer {
    override fun registerStompEndpoints(registry: StompEndpointRegistry) {
        registry.addEndpoint("/portfolio")
    }

    override fun configureMessageBroker(registry: MessageBrokerRegistry) {
        registry.setApplicationDestinationPrefixes("/app")
        registry.enableSimpleBroker("/topic")
    }
}

kotlin

Java

@Controller
public class GreetingController {

    @MessageMapping("/greeting")
    public String handle(String greeting) {
        return "[" + getTimestamp() + ": " + greeting;
    }

    private String getTimestamp() {
        return new SimpleDateFormat("MM/dd/yyyy h:mm:ss a").format(new Date());
    }

}

java

Kotlin

@Controller
class GreetingController {
    
    @MessageMapping("/greeting")
    fun handle(greeting: String): String {
        return "[${getTimestamp()}: $greeting"
    }

    private fun getTimestamp(): String {
        return SimpleDateFormat("MM/dd/yyyy h:mm:ss a").format(Date())
    }
}

kotlin

前面的示例支持以下流程：

1. 客户端连接到 [localhost:8080/portfolio](http://localhost:8080/portfolio)，一旦建立了 WebSocket 连接，STOMP 帧开始在其上流动。

2. 客户端发送一个带有目标头 /topic/greeting 的 SUBSCRIBE 帧。一旦接收到并解码，消息被发送到 clientInboundChannel，然后路由到消息代理，消息代理存储客户端的订阅。

3. 客户端发送一个 SEND 帧到 /app/greeting。/app 前缀有助于将其路由到注解控制器。在去掉 /app 前缀后，剩余的 /greeting 部分的目标被映射到 GreetingController 中的 @MessageMapping 方法。

4. 从 GreetingController 返回的值被转换为一个 Spring Message，其有效载荷基于返回值，并具有默认目标头 /topic/greeting（从输入目标中将 /app 替换为 /topic 得到）。生成的消息被发送到 brokerChannel，并由消息代理处理。

5. 消息代理找到所有匹配的订阅者，并通过 clientOutboundChannel 向每个订阅者发送一个 MESSAGE 帧，从那里消息被编码为 STOMP 帧并通过 WebSocket 连接发送。

下一节提供了有关注释方法的更多细节，包括支持的参数类型和返回值。