Real-time ChatReal-time Chat

kết nối đồng thời lúc cao điểmconcurrent connections at peak

10M

kết nối đồng thời lúc cao điểmconcurrent connections at peak

kết nối mỗi gateway nodeconnections per gateway node

~50k

kết nối mỗi gateway nodeconnections per gateway node

tin nhắn lúc cao điểmmessages per second at peak

60k/s

tin nhắn lúc cao điểmmessages per second at peak

từ người gửi tới người nhận onlinesender to online recipient

p99 < 300 ms

từ người gửi tới người nhận onlinesender to online recipient

Client giữ WebSocket tới một gateway bất kỳ sau load balancer. Chat Service ghi tin vào DB trước, cấp seq theo hội thoại, rồi tra Redis xem người nhận đang ở gateway nào và publish vào kênh của gateway đó. Bật “B offline” để xem push và bù tin khi kết nối lại, hoặc “Gửi lại sau lỗi mạng” để xem chống trùng.Clients hold a WebSocket to any gateway behind the load balancer. The Chat Service stores the message first, assigns a per-conversation seq, looks up in Redis which gateway holds the recipient and publishes to that gateway's channel. Turn on “B is offline” to see the push and the catch-up on reconnect, or “Resend after a network error” to see deduplication.

1. 1

   Định tuyến giữa các gatewayRouting between gateways

   Người gửi và người nhận thường ở hai node khác nhau. Lưu presence:{user} → gateway_id trong Redis; mỗi gateway subscribe kênh gw:{id} của riêng nó. Chat Service đóng vai mediator: client không bao giờ nói chuyện trực tiếp với nhau.Sender and recipient usually sit on different nodes. Keep presence:{user} → gateway_id in Redis; each gateway subscribes to its own gw:{id} channel. The Chat Service acts as the mediator: clients never talk to each other directly.

   Publish-SubscribeMediator
2. 2

   Thứ tự tin trong hội thoạiOrdering within a conversation

   Timestamp của client hay của nhiều server đều lệch. Cấp seq tăng dần theo conversation dưới khoá dòng (hoặc một writer duy nhất cho mỗi partition Kafka key = conversation_id). Client sắp theo seq và phát hiện lỗ hổng để gọi sync.Client clocks and server clocks both drift. Assign an increasing seq per conversation under a row lock (or a single writer per Kafka partition keyed by conversation_id). Clients sort by seq and detect gaps to trigger a sync.

   Producer-Consumer
3. 3

   Gửi trùng khi client retryDuplicates when the client retries

   Ack có thể mất trên đường về. Client giữ client_msg_id cho tới khi nhận ack; server có unique(conversation_id, client_msg_id) và trả lại seq cũ thay vì ghi mới. Client retry với backoff + jitter để cả nghìn máy không cùng kết nối lại một nhịp sau khi gateway khởi động lại.An ack can be lost on the way back. The client keeps its client_msg_id until acked; the server has unique(conversation_id, client_msg_id) and returns the old seq instead of a new row. Clients retry with backoff and jitter so thousands of devices do not reconnect in lockstep after a gateway restart.

   Retry with Exponential Backoff
4. 4

   Đã nhận, đã xemDelivered and read receipts

   Không lưu một dòng cho mỗi tin × mỗi người. Lưu last_delivered_seq và last_read_seq cho mỗi thành viên; mọi tin có seq ≤ giá trị đó là đã nhận / đã xem. Thay đổi được phát cho người gửi như một sự kiện.Do not store one row per message per member. Store last_delivered_seq and last_read_seq per member; every message with seq ≤ that value is delivered or read. Changes are pushed to the sender as events.

   Observer
5. 5

   Offline và bù tinOffline users and catch-up

   Không giữ hàng đợi riêng cho từng người trên gateway. DB là nguồn sự thật; khi kết nối lại, client gửi seq cuối cho từng hội thoại và server trả theo trang. Push qua FCM/APNs chỉ để đánh thức, gộp theo collapse-id.Do not keep a per-user queue on the gateway. The DB is the source of truth; on reconnect the client sends its last seq per conversation and the server pages through the rest. FCM/APNs push only wakes the user and is collapsed by conversation.

   Producer-ConsumerWorker Pool
6. 6

   Client chậm và backpressureSlow clients and backpressure

   Một máy 3G yếu không được làm đầy RAM của gateway. Mỗi socket có hàng gửi giới hạn (ví dụ 256 frame) và write deadline 10 giây; đầy thì đóng socket, client kết nối lại và sync theo seq.A phone on weak 3G must not fill the gateway's memory. Each socket gets a bounded send buffer (say 256 frames) and a 10-second write deadline; when it is full the socket is closed and the client reconnects and syncs by seq.

   Producer-Consumer

Java: Chat Service ghi tin dưới khoá conversation, chống trùng bằng client_msg_id và chỉ định tuyến sau khi commit. Go: gateway giữ hàng gửi có giới hạn cho mỗi socket, ping định kỳ và gia hạn presence.Java: the Chat Service stores the message under the conversation lock, deduplicates by client_msg_id and routes only after commit. Go: the gateway keeps a bounded send buffer per socket, pings regularly and refreshes presence.

@Service
@RequiredArgsConstructor
public class MessageService {
    private final MessageRepository messages;
    private final ConversationRepository conversations;
    private final PresenceRegistry presence;
    private final StringRedisTemplate redis;
    private final PushPublisher push;

    @Transactional
    public Ack send(String senderId, SendCommand cmd) {
        long last = conversations.lockLastSeq(cmd.conversationId()); // FOR UPDATE1
        Optional<Message> dup = messages
            .findByConversationIdAndClientMsgId(cmd.conversationId(), cmd.clientMsgId());
        if (dup.isPresent()) return Ack.of(dup.get(), true);         // retry: same seq2

        long seq = last + 1;
        conversations.updateLastSeq(cmd.conversationId(), seq);
        Message m = messages.save(Message.of(cmd, senderId, seq));
        afterCommit(() -> route(m));3
        return Ack.of(m, false);
    }

    private void route(Message m) {
        for (String userId : conversations.memberIds(m.conversationId())) {
            if (userId.equals(m.senderId())) continue;
            Optional<String> gw = presence.gatewayOf(userId);      // GET presence:{user}
            if (gw.isPresent()) {
                redis.convertAndSend("gw:" + gw.get(), Envelope.json(userId, m));4
            } else {
                push.enqueue(userId, m);                           // offline: FCM / APNs
            }
        }
    }

    private static void afterCommit(Runnable r) {
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronization() {
                @Override public void afterCommit() { r.run(); }
            });
    }
}

type Client struct {
	userID string
	conn   *websocket.Conn
	send   chan []byte // bounded: this is the backpressure1
}

const (
	sendBuffer  = 256
	writeWait   = 10 * time.Second
	pingPeriod  = 25 * time.Second
	presenceTTL = 60 * time.Second
)

// Deliver is called by the subscriber of channel gw:{thisGatewayID}.
func (h *Hub) Deliver(userID string, frame []byte) {
	h.mu.RLock()
	c, ok := h.clients[userID]
	h.mu.RUnlock()
	if !ok {
		return // user moved away; catch-up by seq covers it
	}
	select {
	case c.send <- frame:
	default:2
		c.conn.Close() // slow consumer: drop it, the client resyncs
	}
}

func (c *Client) writePump(ctx context.Context, p Presence, gatewayID string) {
	ticker := time.NewTicker(pingPeriod)
	defer ticker.Stop()
	for {
		select {
		case frame := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))3
			if err := c.conn.WriteMessage(websocket.TextMessage, frame); err != nil {
				return
			}
		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
			p.Set(ctx, c.userID, gatewayID, presenceTTL) // SET presence:{u} gw EX 604
		case <-ctx.Done():
			return
		}
	}
}

Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.