如何使用 RxStompService 获取 Angular 8 上消息的最新偏移量？

Question

我有订阅 Kafka 主题的 RxStompService，但是当我从 UI 端更新数据并通过 REST 调用将其发送到后端时，它会显示消息的所有偏移量，这意味着重复消息。有没有办法在订阅时显示消息的最新偏移量？

这是我的 UI 代码；

    ngOnInit(): void {
        this.rxStompService.watch('/topic/notification')
          .subscribe(value => {
            console.log('Received from websocket: ' + value.body);
            this.rows.push({
              outageId: Number(JSON.parse(value.body).outageId),
              stage: JSON.parse(value.body).stage,
              startTime: JSON.parse(value.body).timestamp == undefined ? new Date(parseInt(JSON.parse(value.body).startTime, 10)) : new Date(JSON.parse(value.body).timestamp),
              endTime: JSON.parse(value.body).timestamp == undefined ? (JSON.parse(value.body).endTime == '-' ? '-' : new Date(parseInt(JSON.parse(value.body).endTime, 10)))
                : (JSON.parse(value.body).stage == 'CLOSED' ? new Date(JSON.parse(value.body).timestamp) : '-'),
              type: JSON.parse(value.body).type,
              details: JSON.parse(value.body).details
            });
            this.rows = this.rows.slice(-100);
          });
      }

还有我的 myRxStompConfig；

    import { InjectableRxStompConfig } from '@stomp/ng2-stompjs';
    
    export const myRxStompConfig: InjectableRxStompConfig = {
        brokerURL: 'ws://localhost:8080/websocket',
    
        connectHeaders: {
            login: '',
            passcode: ''
        },
    
        heartbeatIncoming: 0, // Typical value 0 - disabled
        heartbeatOutgoing: 20000, // Typical value 20000 - every 20 seconds
    
        reconnectDelay: 1000,
    
        debug: (msg: string): void => {
            console.log(new Date(), msg);
        }
    };

Answer 1

据我所知，Apache Kafka 甚至不支持 STOMP，所以我认为您的问题没有实际意义。根据this，Kafka 不支持 STOMP。另外，我在 Kafka 网站上找不到任何对 STOMP 实现的引用。