@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
private final ChatWebSocketHandler chatWebSocketHandler;
private final AuthHandshakeInterceptor authHandshakeInterceptor;
public WebSocketConfig(ChatWebSocketHandler chatWebSocketHandler,
AuthHandshakeInterceptor authHandshakeInterceptor) {
this.chatWebSocketHandler = chatWebSocketHandler;
this.authHandshakeInterceptor = authHandshakeInterceptor;
}
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
registry.addHandler(chatWebSocketHandler, "/ws/chat")
.addInterceptors(authHandshakeInterceptor)
.setAllowedOriginPatterns("*");
}
}
握手鉴权：
@Component
public class AuthHandshakeInterceptor implements HandshakeInterceptor {
private final JwtTokenService jwtTokenService;
public AuthHandshakeInterceptor(JwtTokenService jwtTokenService) {
this.jwtTokenService = jwtTokenService;
}
@Override
public boolean beforeHandshake(ServerHttpRequest request,
ServerHttpResponse response,
WebSocketHandler wsHandler,
Map<String, Object> attributes) {
String token = request.getHeaders().getFirst(HttpHeaders.AUTHORIZATION);
if (token == null || !token.startsWith("Bearer ")) {
response.setStatusCode(HttpStatus.UNAUTHORIZED);
return false;
}
UserPrincipal principal = jwtTokenService.parse(token.substring(7));
attributes.put("userId", principal.userId());
attributes.put("deviceId", principal.deviceId());
return true;
}
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler wsHandler, Exception exception) {
}
}
本地连接管理：
@Component
public class SessionManager {
private final ConcurrentMap<Long, CopyOnWriteArraySet<WebSocketSession>> userSessions =
new ConcurrentHashMap<>();
public void register(Long userId, WebSocketSession session) {
userSessions.computeIfAbsent(userId, k -> new CopyOnWriteArraySet<>()).add(session);
}
public void unregister(Long userId, WebSocketSession session) {
CopyOnWriteArraySet<WebSocketSession> sessions = userSessions.get(userId);
if (sessions == null) {
return;
}
sessions.remove(session);
if (sessions.isEmpty()) {
userSessions.remove(userId);
}
}
public boolean isOnlineLocally(Long userId) {
CopyOnWriteArraySet<WebSocketSession> sessions = userSessions.get(userId);
return sessions != null && !sessions.isEmpty();
}
public void sendToUser(Long userId, String payload) {
CopyOnWriteArraySet<WebSocketSession> sessions = userSessions.get(userId);
if (sessions == null) {
return;
}
for (WebSocketSession session : sessions) {
if (session.isOpen()) {
try {
session.sendMessage(new TextMessage(payload));
} catch (IOException ignored) {
}
}
}
}
}
网关只做轻逻辑：
@Component
public class ChatWebSocketHandler extends TextWebSocketHandler {
private final SessionManager sessionManager;
private final PresenceService presenceService;
private final UpstreamEventPublisher upstreamEventPublisher;
private final ObjectMapper objectMapper;
public ChatWebSocketHandler(SessionManager sessionManager,
PresenceService presenceService,
UpstreamEventPublisher upstreamEventPublisher,
ObjectMapper objectMapper) {
this.sessionManager = sessionManager;
this.presenceService = presenceService;
this.upstreamEventPublisher = upstreamEventPublisher;
this.objectMapper = objectMapper;
}
@Override
public void afterConnectionEstablished(WebSocketSession session) {
Long userId = (Long) session.getAttributes().get("userId");
sessionManager.register(userId, session);
presenceService.online(userId);
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
Long userId = (Long) session.getAttributes().get("userId");
WsPacket packet = objectMapper.readValue(message.getPayload(), WsPacket.class);
switch (packet.type()) {
case CHAT_SEND -> upstreamEventPublisher.publishChatSend(userId, packet);
case READ_ACK -> upstreamEventPublisher.publishReadAck(userId, packet);
case PING -> presenceService.renew(userId);
default -> throw new IllegalArgumentException("unsupported packet type");
}
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
Long userId = (Long) session.getAttributes().get("userId");
sessionManager.unregister(userId, session);
presenceService.offlineIfNoSession(userId);
}
}
这段代码有一个很重要的生产原则：网关不直接写业务库，不直接推进已读，不直接计算未读。
它只负责：
1）接包；
2）轻量校验；
3）丢进 Kafka；
4）维护连接。
9、消息发送链路一次讲透
发送事件模型：
public record ChatSendEvent(
String eventId,
Long conversationId,
Long senderId,
Long receiverId,
String clientMsgId,
String msgType,
String content,
Instant occurredAt
) {
}
Kafka 生产：
@Component
public class UpstreamEventPublisher {
private final KafkaTemplate<String, Object> kafkaTemplate;
public UpstreamEventPublisher(KafkaTemplate<String, Object> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void publishChatSend(Long userId, WsPacket packet) {
ChatSendEvent event = new ChatSendEvent(
UUID.randomUUID().toString(),
packet.conversationId(),
userId,
packet.peerUserId(),
packet.clientMsgId(),
packet.msgType(),
packet.content(),
Instant.now()
);
kafkaTemplate.send("chat.send", String.valueOf(packet.conversationId()), event);
}
public void publishReadAck(Long userId, WsPacket packet) {
ReadAckEvent event = new ReadAckEvent(
UUID.randomUUID().toString(),
packet.conversationId(),
userId,
packet.ackSeq(),
Instant.now()
);
kafkaTemplate.send("receipt.ack", String.valueOf(packet.conversationId()), event);
}
}
这里选择 conversationId 作为分区键，是因为：
1）同一会话内消息必须尽量保持顺序；
2）已读 ACK 也希望进入同一分区，降低乱序复杂度。
消息持久化服务：
@Service
public class ChatCommandService {
private final ConversationRepository conversationRepository;
private final MessageRepository messageRepository;
private final UnreadCacheService unreadCacheService;
private final PushEventPublisher pushEventPublisher;
@Transactional
public void handle(ChatSendEvent event) {
if (messageRepository.existsByClientMsgId(event.clientMsgId())) {
return;
}
Conversation conversation = conversationRepository.lockById(event.conversationId())
.orElseThrow();
long nextSeq = conversation.getLastMsgSeq() + 1;
conversation.setLastMsgSeq(nextSeq);
conversationRepository.save(conversation);
Message message = Message.builder()
.conversationId(event.conversationId())
.msgSeq(nextSeq)
.senderId(event.senderId())
.receiverId(event.receiverId())
.msgType(event.msgType())
.content(event.content())
.clientMsgId(event.clientMsgId())
.sendTime(LocalDateTime.now())
.build();
messageRepository.save(message);
unreadCacheService.increment(event.receiverId(), event.conversationId(), 1L);
pushEventPublisher.publishMessage(message);
}
}
这里用了一个常见但很有效的做法：对会话行加锁，串行分配 msg_seq。
对于单聊来说，会话维度天然足够细，这种做法比全局序列更简单，也更容易保证顺序。
如果未来要进一步扩展到极高吞吐，可把会话序号分配升级为：
1）Redis 原子自增；
2）Snowflake + 会话内重排；
3）号段分配器。
但大多数中大型业务，先把正确性做稳，比过早设计复杂序列更重要。
10、已读推进链路：用Lua保证原子性与幂等
ACK 事件：
public record ReadAckEvent(
String eventId,
Long conversationId,
Long userId,
Long ackSeq,
Instant occurredAt
) {
}
Redis Lua 脚本：
local key = KEYS[1]
local field = ARGV[1]
local newSeq = tonumber(ARGV[2])
local oldSeq = redis.call('HGET', key, field)
if oldSeq then
oldSeq = tonumber(oldSeq)
else
oldSeq = 0
end
if newSeq <= oldSeq then
return {oldSeq, oldSeq}
end
redis.call('HSET', key, field, newSeq)
return {oldSeq, newSeq}
已读服务实现：
@Service
public class ReadReceiptService {
private final StringRedisTemplate redisTemplate;
private final ReadCursorRepository readCursorRepository;
private final UnreadCacheService unreadCacheService;
private final PushEventPublisher pushEventPublisher;
private final DefaultRedisScript<List> advanceCursorScript;
@Transactional
public void handle(ReadAckEvent event) {
String key = "im:read:cursor:" + event.conversationId();
List<Long> result = redisTemplate.execute(
advanceCursorScript,
Collections.singletonList(key),
String.valueOf(event.userId()),
String.valueOf(event.ackSeq())
);
long oldSeq = result.get(0);
long newSeq = result.get(1);
if (newSeq <= oldSeq) {
return;
}
unreadCacheService.decreaseByCursor(event.userId(), event.conversationId(), oldSeq, newSeq);
readCursorRepository.upsert(event.conversationId(), event.userId(), newSeq, LocalDateTime.now());
pushEventPublisher.publishReadNotify(event.conversationId(), event.userId(), newSeq);
}
}
@Service
public class PushRouterService {
private final PresenceService presenceService;
private final KafkaTemplate<String, Object> kafkaTemplate;
public void routeMessage(Long userId, Object payload) {
String gatewayId = presenceService.findGateway(userId);
if (gatewayId == null) {
return;
}
GatewayPushEvent event = new GatewayPushEvent(gatewayId, userId, payload);
kafkaTemplate.send("gateway.push", gatewayId, event);
}
}
网关消费定向推送：
@KafkaListener(topics = "gateway.push", groupId = "ws-gateway")
public void onGatewayPush(GatewayPushEvent event) {
if (!localInstanceId.equals(event.gatewayId())) {
return;
}
sessionManager.sendToUser(event.userId(), toJson(event.payload()));
}
注意这里的消费模型要根据你的 Kafka group 策略设计：
1）如果一个网关实例对应一个独立消费者身份，可以精准消费；
2）如果所有网关共用一个 group，则要确保 topic 分区与实例映射清晰。
很多团队这里会踩坑。
我的建议是：
1）中小规模：Redis Pub/Sub 或广播消费就够用；
2）中大规模：独立 push-router + Kafka 定向 topic 更稳。
12、未读数缓存不是事实源，修正机制必须存在
首页会话列表、消息角标、应用角标都会频繁读取未读数，所以未读数必须走缓存。
但缓存未读数很容易在以下场景中漂移：
1）消费失败导致少加一次；
2）ACK 重试导致多减一次；
3）补偿任务重复回放；
4）节点故障导致中间状态未提交。
因此，生产设计必须遵循一个原则：未读数可以短暂不一致，但必须可重算、可修正。
推荐做法：
1）Redis 维护热未读数，服务接口优先读 Redis；
2）MySQL 持久化已读游标与最后消息序号；
3）定时任务对热点会话做校准。
Redis 未读数接口：
@Service
public class UnreadCacheService {
private final StringRedisTemplate redisTemplate;
public void increment(Long userId, Long conversationId, Long delta) {
redisTemplate.opsForHash().increment(
"im:unread:" + userId,
String.valueOf(conversationId),
delta
);
}
public void decreaseByCursor(Long userId, Long conversationId, long oldSeq, long newSeq) {
long delta = Math.max(0, newSeq - oldSeq);
if (delta == 0) {
return;
}
redisTemplate.opsForHash().increment(
"im:unread:" + userId,
String.valueOf(conversationId),
-delta
);
}
}
默认
首页
