AngularJS的websockets太快了

Question

我正在使用AngularJS来驱动我的web应用程序。最近，我决定在RabbitMQ中引入Stomp over websockets。我一直遇到的一个问题是，有时从后端发送的消息频率太高，以至于Angular无法触发一些事件，在这个过程中丢失了数据。

应用程序相当复杂，因此有时消息可能会通过以下链

StompJS Angular服务包装器。使用rootScope美元。$broadcast当新消息到达时通知组件
1. 注册在$rootScope上广播的消息的控制器。然后控制器更新绑定或发送另一个事件到组件来执行一些数据更新。

正如我提到的有时作用域没有得到正确的更新，即使正在发送事件并且正在填充值，视图也没有显示更新的值。我使用了$scope。$apply， $timeout， $scope。$digest和似乎没有工作…例如，如果我有两个数据包一个接一个地来，它们之间在套接字上几乎没有延迟，什么也不会发生，但如果有一个相同的代码段正常运行。如何克服这个问题?

在用户提出的一些问题之后进一步举例:

我举一个最简单的例子:我从后端运行的作业中获取进度报告。基本上，它告诉我在Cassandra数据库中写了多少行。所以我得到像{written: 8000, total: 4000000}这样的通知。随着工作人员在数据库中写入数据，"写"的数量也在增加，而且由于有多个工作人员，有时这些通知被推送得非常快。我编写了一个自定义网格组件，用于监听grid:updateCell(允许我使用进度报告更新单元格)等事件。每当一个数据包到达套接字时，我就广播该事件($scope)。$broadcast(因为网格是控制器的子页面)。我注意到，并非所有来自套接字的更新都反映在UI中，尽管事件被捕获，网格事件也被触发，数据模型的成功更新，但不是UI

Answer 1

在我看来，这里可能有一些事情正在进行。

1)使用$scope。如果你有很多嵌套的作用域，那么$broadcast会显示作用域，这可能会很重。我更喜欢将事件监听器附加到$rootScope，并使用$emit，它只广播到当前范围并且更快。我有以下服务

    /**
     * @ngdoc service
     * @name someModule.eventbus
     * @requires $rootScope
     *
     * @description
     * Provides a eventing mechanism when a user cna broadcast and subscribe to application wide events.
     */
    angular.module('someModule').factory('eventbus', [
        '$rootScope',
        function ($rootScope) {
            /**
             * @ngdoc function
             * @name subscribe
             * @methodOf someModule.eventbus
             *
             * @description
             * Subscribes a callback to the given application wide event
             *
             * @param {String} eventName The name of the event to subscribe to.
             * @param {Function} callback A callback which is fire when the event is raised.
             * @return {Function} A function tht can be called to unsubscrive to the event.
             */
            var subscribe = function (eventName, callback) {
                    return $rootScope.$on(eventName, callback);
                },
                /**
                 * @ngdoc function
                 * @name broadcast
                 * @methodOf someModule.eventbus
                 *
                 * @description
                 * Broadcasts the given event and data.
                 *
                 * @param {String} eventName The name of the event to broadcast.
                 * @param {object} data A data object that will be passed along with the event.
                 */
                broadcast = function (eventName, data) {
                    $rootScope.$emit(eventName, data);
                };
            return {
                subscribe: subscribe,
                broadcast: broadcast
            };
        }
    ]);

1. 你可能没有在Angular的摘要周期内触发更新，这就解释了为什么你只获得了一些更新。尝试使用摘要循环进行更新(将函数放在$timeout块中，这是在$scope上的推荐方法。

   $timeout(function () {doUpdate (dataFromMessageBus);});

2. 如果你收到很多消息，你应该使用一个debounce函数，参见lodash-debouce。一个人实际上只能处理大约200ms的数据块，所以一般来说，每200ms更新一次是一个人能接受的。我个人每隔一秒左右就会在交易应用程序等中执行一次。

(使用lodash - debounce)

function (messageBus) {
  var debounceFn = _.debounce(function () {
      $timeout(function() {
        doUpdate(data);
      });
    }, 1000);
    // just assuming here about your socket / message bus stuff
    messageBus.onMessageRecieved(function () {
      //Call debounce function which will only actually call the inner func every 1000ms
      debounceFn();
   });
}

Answer 2

我以前有过同样的问题，我能够通过包装websocket来解决它传入消息处理程序，它更新我的Angular Grid如下:

$scope.$apply(function () {
//your code to update the Angular UI
});

的例子:

    /**
     * @ngdoc function
     * @description
     * On Connect. Callback to subscribe to specific socket destination. 
     */
    function _subscribeBcQueue(){
        var uuid = SessionService.getUuid();
        var userBcQueue = CONFIG.SOCKET.bcQueue+'/'+uuid
        var subscription = $stomp.subscribe(userBcQueue, function (payload, headers, message) {
            _onBcMessageHandler(message)
        }, {})
    }
    /**
     * @ngdoc function
     * @description
     * OnBcMessageHandler. Callback function to handle the incoming direct message. 
     */
    function _onBcMessageHandler(message){
        //TODO Process the broadcasting status message
        var bcMsg = JSON.parse(message.body);
        console.log("Broadcasting status message:" + bcMsg);
        _updateBcJobGridRow(bcMsg)
    }

    function _updateBcJobGridRow(job) {
            $scope.$apply(function () {
                //Get the old JobGridRow 
                console.log("jobId: " + JSON.stringify(job.id));
                var oldJobGridRow = $filter('filter')(vm.gridOptions.data, {id: job.id})[0];
                console.log("oldJobGridRow: " + JSON.stringify(oldJobGridRow));
                if (oldJobGridRow) {
                    var newJobGridRow = _getBcJobGridRow(job);
                    console.log("newJobGridRow: " + JSON.stringify(newJobGridRow));
                    oldJobGridRow.progress = newJobGridRow.progress;
                    oldJobGridRow.percentage = newJobGridRow.percentage;
                    oldJobGridRow.status = newJobGridRow.status;
                    oldJobGridRow.total = newJobGridRow.total;
                    oldJobGridRow.sent = newJobGridRow.sent;
                    oldJobGridRow.recipients = newJobGridRow.recipients;
                }
            });
        }