我正在使用 SimpUserRegistry
获取在线用户数(使用 getUserCount()
)。它在我的本地机器上运行良好,但在只有弹性 IP 且没有负载均衡器的 AWS EC2 实例(尝试使用 Amazon Linux 和 Ubuntu)上运行良好。
EC2 上的问题是某些用户在连接时从未添加到注册表中,因此我得到了错误的结果。
我有 session 监听器,用于 SessionConnectedEvent
和 SessionDisconnectEvent
,我在其中使用 SimpUserRegistry
(自动连接)来获取用户存在。如果重要的话,我也是 SimpUserRegistry
是一个消息传递 Controller 。
下面是 websocket 消息代理配置:
@Order(Ordered.HIGHEST_PRECEDENCE + 99)
@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class WebSocketMessageBrokerConfig extends AbstractWebSocketMessageBrokerConfigurer {
@NonNull
private SecurityChannelInterceptor securityChannelInterceptor;
@Override
public void configureMessageBroker(MessageBrokerRegistry config) {
ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
threadPoolTaskScheduler.setPoolSize(1);
threadPoolTaskScheduler.setThreadGroupName("cb-heartbeat-");
threadPoolTaskScheduler.initialize();
config.enableSimpleBroker("/queue/", "/topic/")
.setTaskScheduler(threadPoolTaskScheduler)
.setHeartbeatValue(new long[] {1000, 1000});
config.setApplicationDestinationPrefixes("/app");
}
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/websocket")
.setAllowedOrigins("*")
.withSockJS();
}
@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
registration.interceptors(securityChannelInterceptor);
}
}
下面是上面配置类中使用的 channel 拦截器:
@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class SecurityChannelInterceptor extends ChannelInterceptorAdapter {
@NonNull
private SecurityService securityService;
@Value("${app.auth.token.header}")
private String authTokenHeader;
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
StompCommand command = accessor.getCommand();
if (StompCommand.CONNECT.equals(command)) {
List<String> authTokenList = accessor.getNativeHeader(authTokenHeader);
if (authTokenList == null || authTokenList.isEmpty()) {
throw new AuthenticationFailureException("STOMP " + command + " missing " + this.authTokenHeader + " header!");
}
String accessToken = authTokenList.get(0);
AppAuth authentication = securityService.authenticate(accessToken);
log.info("STOMP {} authenticated. Authentication Token = {}", command, authentication);
accessor.setUser(authentication);
SecurityContextHolder.getContext().setAuthentication(authentication);
Principal principal = accessor.getUser();
if (principal == null) {
throw new RuntimeException("StompHeaderAccessor did not set the authenticated User for " + authentication);
}
}
return message;
}
}
我还有以下计划任务,它只是每两秒打印一次用户名:
@Component
@Slf4j
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class UserRegistryLoggingTask {
private SimpUserRegistry simpUserRegistry;
@Scheduled(fixedRate = 2000)
public void logUsersInUserRegistry() {
Set<String> userNames = simpUserRegistry.getUsers().stream().map(u -> u.getName()).collect(Collectors.toSet());
log.info("UserRegistry has {} users with IDs {}", userNames.size(), userNames);
}
}
有些用户名即使在连接后也不会显示。
SecurityService
类的实现-
@Service
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class SecurityService {
private UserRepository userRepository;
private UserCredentialsRepository userCredentialsRepository;
private JwtHelper jwtHelper;
public User getUser() {
AppAuth auth = (AppAuth) SecurityContextHolder.getContext().getAuthentication();
User user = (User) auth.getUser();
return user;
}
public AppAuth authenticate(String accessToken) {
String username = jwtHelper.tryExtractSubject(accessToken);
if (username == null) {
throw new AuthenticationFailureException("Invalid access token!");
}
User user = userRepository.findByUsername(username);
if (user == null) {
throw new AuthenticationFailureException("Invalid access token!");
}
AppAuth authentication = new AppAuth(user);
return authentication;
}
}
更新
以下是 SockJS 在浏览器上的日志示例 -
来自带有 user-name
header 的服务器的正确响应:
>>> CONNECT
AccessToken:eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJkb2cifQ.Wf8AO77LluHEfEv61TIvugEXxOqIXKjsJBO8QMQh-rF7tzf56lBkdpOruqc7UPf_Pmj6-dnHZ5raq2MnMpeG8Q
accept-version:1.1,1.0
heart-beat:10000,10000
<<< CONNECTED
version:1.1
heart-beat:1000,1000
user-name:5a590e411b96f841cc00027f
来自没有user-name
header 的服务器的错误响应:
>>> CONNECT
AccessToken:eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJtb3VzZSJ9.wqX5X_CSdHD8_7PZPiSzftGCuPz1ClQU0-F9RHCqOIIkMLzI4rt31_EAaykc8VojK2KGS6DcycWfAdMr2edzYg
accept-version:1.1,1.0
heart-beat:10000,10000
<<< CONNECTED
version:1.1
heart-beat:1000,1000
我还验证了 SecurityChannelInterceptor
正在验证所有用户,即使 user-name
不在 CONNECTED
响应中也是如此。
更新
我在 heroku 上部署了该应用程序。问题也在那里发生。
更新
发生问题时,SessionConnectEvent
中的user
是SecurityChannelInterceptor
设置的,但 中的
为 user
SessionConnectedEventnull
。
更新
AppAuth
类 -
public class AppAuth implements Authentication {
private final User user;
private final Collection<GrantedAuthority> authorities;
public AppAuth(User user) {
this.user = user;
this.authorities = Collections.singleton((GrantedAuthority) () -> "USER");
}
public User getUser() {
return this.user;
}
@Override
public String getName() {
return user.getId();
}
@Override
public Collection<? extends GrantedAuthority> getAuthorities() {
return authorities;
}
@Override
public Object getCredentials() {
return null;
}
@Override
public Object getDetails() {
return null;
}
@Override
public Object getPrincipal() {
return new Principal() {
@Override
public String getName() {
return user.getId();
}
};
}
@Override
public boolean isAuthenticated() {
return true;
}
@Override
public void setAuthenticated(boolean isAuthenticated) throws IllegalArgumentException {
}
}
最佳答案
通过在 StompSubProtocolHandler
中添加一些记录器语句,我能够在一些调试后跟踪问题。
找到原因后,结论是 channel 拦截器不是对用户进行身份验证的正确位置。至少对于我的用例而言。
以下是来自 StompSubProtocolHandler
的一些代码片段-
handleMessageFromClient
方法将用户添加到 stompAuthentications
映射并发布一个 SessionConnectEvent
事件 -
public void handleMessageFromClient(WebSocketSession session, WebSocketMessage<?> webSocketMessage, MessageChannel outputChannel) {
//...
SimpAttributesContextHolder.setAttributesFromMessage(message);
boolean sent = outputChannel.send(message);
if (sent) {
if (isConnect) {
Principal user = headerAccessor.getUser();
if (user != null && user != session.getPrincipal()) {
this.stompAuthentications.put(session.getId(), user);
}
}
if (this.eventPublisher != null) {
if (isConnect) {
publishEvent(new SessionConnectEvent(this, message, getUser(session)));
}
//...
handleMessageToClient
从 stompAuthentications
映射中检索用户并发布一个 SessionConnectedEvent
-
public void handleMessageToClient(WebSocketSession session, Message<?> message) {
//...
SimpAttributes simpAttributes = new SimpAttributes(session.getId(), session.getAttributes());
SimpAttributesContextHolder.setAttributes(simpAttributes);
Principal user = getUser(session);
publishEvent(new SessionConnectedEvent(this, (Message<byte[]>) message, user));
//...
getUser
上述方法使用的方法-
private Principal getUser(WebSocketSession session) {
Principal user = this.stompAuthentications.get(session.getId());
return user != null ? user : session.getPrincipal();
}
现在,当 handleMessageToClient
片段在 handleMessageFromClient
片段之前执行时,就会出现问题。在这种情况下,用户永远不会添加到 DefaultSimpUserRegistry
,因为它只检查 SessionConnectedEvent
。
下面是来自 DefaultSimpUserRegistry
的事件监听器片段-
public void onApplicationEvent(ApplicationEvent event) {
//...
else if (event instanceof SessionConnectedEvent) {
Principal user = subProtocolEvent.getUser();
if (user == null) {
return;
}
//...
解决方案
解决方案是扩展DefaultHandshakeHandler
并覆盖基于 this answer 的 determineUser
方法.但是,由于我使用的是 SockJS,这需要客户端将 auth-token 作为查询参数发送。并讨论了查询参数要求的原因here .
关于java - 为什么 SimpUserRegistry 在 EC2 实例上无法正常工作,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48246245/