Real-time TrackingReal-time Tracking

tài xế online cùng lúcdrivers online at once

200k

tài xế online cùng lúcdrivers online at once

điểm GPS (mỗi 4 giây một điểm)GPS points (one every 4 seconds)

50k/s

điểm GPS (mỗi 4 giây một điểm)GPS points (one every 4 seconds)

p99 từ điện thoại tài xế tới bản đồ kháchp99 from the driver's phone to the rider's map

< 1 s

p99 từ điện thoại tài xế tới bản đồ kháchp99 from the driver's phone to the rider's map

điểm mỗi ngày trước downsamplingpoints a day before downsampling

~4.3 tỷ

điểm mỗi ngày trước downsamplingpoints a day before downsampling

Điểm GPS đi qua một pipeline: broker MQTT nhận và giới hạn tốc độ, Kafka giữ thứ tự theo tài xế, processor loại điểm cũ rồi tách ra ba nhánh: vị trí mới nhất trong Redis GEO, lịch sử trong TimescaleDB và cập nhật cho người đang xem qua pub/sub. Bật “Điểm đến muộn” và “Tìm xe gần” để xem hai luồng còn lại.GPS points travel a pipeline: an MQTT broker receives and throttles, Kafka keeps per-driver order, and the processor drops stale points and splits into three branches: the latest position in Redis GEO, history in TimescaleDB, and updates to watchers over pub/sub. Turn on “Late points” and “Nearby search” to see the other flows.

1. 1

   50 000 lần ghi mỗi giây50,000 writes a second

   Không ghi từng điểm vào Postgres. Kafka làm bộ đệm, partition theo driver_id; Redis chỉ giữ một dòng cho mỗi tài xế; lịch sử ghi theo lô bằng COPY. Một instance Redis xử lý khoảng 100k lệnh/giây, nên tách cluster theo thành phố.Never write each point to Postgres. Kafka buffers, partitioned by driver_id; Redis keeps one row per driver; history is written in batches with COPY. One Redis instance handles about 100k commands a second, so shard the cluster by city.

   PipelineProducer-Consumer
2. 2

   Điểm cũ và điểm trùngStale and duplicate points

   Mạng di động giao muộn và giao lại. So ts (và seq) với điểm mới nhất trong cùng script Lua để đọc-so-ghi là nguyên tử; điểm cũ bị loại khỏi luồng live nhưng vẫn vào lịch sử. Dùng ts của GPS, không dùng giờ server nhận.Mobile networks deliver late and deliver twice. Compare ts (and seq) with the latest point inside one Lua script so read-compare-write is atomic; stale points leave the live path but still go to history. Use the GPS ts, not the server's receive time.

   Pipeline
3. 3

   Tài xế offline vẫn nằm trong GEOOffline drivers linger in GEO

   GEOADD lưu vào một sorted set, từng thành viên không có TTL. Giữ thêm seen:{city} (ZSET theo last_seen), job mỗi 10 giây ZRANGEBYSCORE rồi ZREM những tài xế im lặng quá 60 giây; truy vấn cũng lọc theo last_seen.GEOADD stores into a sorted set, and members have no TTL. Keep seen:{city} as a ZSET by last_seen; a job every 10 seconds runs ZRANGEBYSCORE and ZREMs drivers silent for over 60 seconds, and queries filter on last_seen too.

   Producer-Consumer
4. 4

   Đẩy cập nhật tới đúng người xemPushing updates to the right watchers

   Không broadcast mọi điểm. Mỗi chuyến một kênh trip:{id}; gateway chỉ subscribe kênh của các WebSocket nó đang giữ. Giới hạn 1 cập nhật mỗi giây cho mỗi người xem, client tự nội suy chuyển động giữa hai điểm.Never broadcast every point. Each trip has a channel trip:{id}; a gateway subscribes only to channels of WebSockets it holds. Cap updates at one a second per watcher and let the client interpolate between points.

   Publish-SubscribeObserverRate Limiter
5. 5

   Lịch sử phình toHistory grows without limit

   4.3 tỷ điểm một ngày. TimescaleDB chia chunk theo giờ, nén chunk cũ hơn 1 ngày (giảm khoảng 90%), continuous aggregate 1 phút cho báo cáo, retention xoá dữ liệu thô sau 7 ngày. Lộ trình chuyến được đơn giản hoá (Douglas-Peucker) và lưu riêng.4.3 billion points a day. TimescaleDB chunks by hour, compresses chunks older than a day (about 90% smaller), keeps a 1-minute continuous aggregate for reporting, and drops raw data after 7 days. Trip routes are simplified (Douglas-Peucker) and stored separately.

   Fan-out / Fan-in
6. 6

   Pin và data của tài xếDriver battery and data

   Tần suất thích ứng: 4 giây khi đang chạy, 15–30 giây khi đứng yên, bỏ điểm dịch dưới 10 m. Gửi protobuf, delta encoding so với điểm trước, gom 3–5 điểm một lần khi không trong chuyến. MQTT giữ một kết nối nhẹ, keepalive 60 giây, thay vì mở HTTPS mỗi lần.Adaptive interval: every 4 seconds while moving, 15–30 seconds when stopped, skip moves under 10 m. Send protobuf with delta encoding against the previous point, and batch 3–5 points when not on a trip. MQTT keeps one light connection with a 60-second keepalive instead of a new HTTPS call each time.

   Flyweight
7. 7

   Dữ liệu lặp lại trong mỗi điểmRepeated data in every point

   Biển số, loại xe, ảnh tài xế không đi kèm từng điểm. Điểm chỉ mang driver_id và toạ độ; thông tin xe là đối tượng dùng chung, client tải một lần khi bắt đầu chuyến và cache lại.Plate, car model and driver photo never travel with each point. A point carries only driver_id and coordinates; vehicle details are a shared object the client loads once at trip start and caches.

   Flyweight

Processor ghi mọi điểm vào lịch sử, còn luồng live chỉ nhận điểm mới hơn điểm đã lưu. Việc so sánh và ghi nằm trong một script Lua để nguyên tử.The processor writes every point to history, while the live path accepts only points newer than the stored one. The compare and the writes live in one Lua script so they are atomic.

@Component
@RequiredArgsConstructor
public class LocationProcessor {
    private static final Duration MAX_LIVE_AGE = Duration.ofSeconds(30);
    private final StringRedisTemplate redis;
    private final RedisScript<Long> updateLatest;      // update_latest.lua
    private final HistoryWriter history;               // COPY into TimescaleDB
    private final Clock clock;

    @KafkaListener(topics = "location.raw", batch = "true", concurrency = "8")1
    public void onBatch(List<LocationPoint> points) {
        history.copy(points);                          // every point, original ts2
        Instant now = clock.instant();
        for (LocationPoint p : points) {
            if (Duration.between(p.ts(), now).compareTo(MAX_LIVE_AGE) > 0) continue;3
            String tag = "{" + p.city() + "}";
            Long applied = redis.execute(updateLatest,
                List.of("drv:" + tag + ":" + p.driverId(), "geo:" + tag, "seen:" + tag),
                p.driverId(), String.valueOf(p.lng()), String.valueOf(p.lat()),
                String.valueOf(p.ts().toEpochMilli()));
            if (applied == null || applied == 0) continue; // older than the latest4
            p.tripId().ifPresent(trip ->
                redis.convertAndSend("trip:" + trip, p.toWireJson()));
        }
    }
}

type Processor struct {
	rdb     *redis.ClusterClient
	latest  *redis.Script // update_latest.lua
	history *HistoryWriter
	maxAge  time.Duration
}

func (p *Processor) Run(ctx context.Context, r *kafka.Reader) error {
	for {
		batch, err := fetchBatch(ctx, r, 500, 200*time.Millisecond)1
		if err != nil {
			return err
		}
		points := decode(batch) // protobuf; malformed points are counted and skipped
		if err := p.history.Copy(ctx, points); err != nil {
			return err // offsets not committed: the batch is read again2
		}
		for i := range points {
			if time.Since(points[i].Time()) <= p.maxAge {
				p.applyLive(ctx, &points[i])
			}
		}
		if err := r.CommitMessages(ctx, batch...); err != nil {
			return err
		}
	}
}

func (p *Processor) applyLive(ctx context.Context, pt *Point) {
	tag := "{" + pt.City + "}"
	keys := []string{"drv:" + tag + ":" + pt.DriverID, "geo:" + tag, "seen:" + tag}3
	ok, err := p.latest.Run(ctx, p.rdb, keys, pt.DriverID, pt.Lng, pt.Lat, pt.TsMillis).Int()
	if err != nil || ok == 0 {
		return // stale point, or a Redis hiccup the next point repairs4
	}
	if pt.TripID != "" {
		p.rdb.Publish(ctx, "trip:"+pt.TripID, pt.WireJSON())
	}
}

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