Nodejs：错误的回调顺序

【问题标题】：Nodejs: wrong callback orderNodejs：错误的回调顺序
【发布时间】：2018-05-11 20:58:23
【问题描述】：

我有 3 个使用 web 套接字交换消息的 nodejs 服务。

第一个服务通过网络套接字将事件发送到第二个。第二个服务负责详细说明事件并将它们保存在 mongodb 数据库中，一旦将事件保存在数据库中，第二个服务始终使用 Web 套接字将详细说明的事件发送到第三个服务（gui）。

这是从第一个接收事件的第二个服务的代码：

this.socketClient.on('NEW_EVENT', (event: MyEvent) => {
 winston.info(`[newEvent] New Event received: %s`, event.type);
 console.log('0__', event.payload);

 // Search for the right class that can handle the specific event
 for (const handler of this.handlers) {
   if (handler.isApplicable(event.type)) {
       // Handler found
       handler.handle(event).then((handledEvent) => {
          console.log('1__', handledEvent);  
          // Send the handled event to the gui
          this.webSockerService.publishNewEvent(event.type, handledEvent);
       });
       return;
  }
 }
});

这是负责处理一个特定事件的 Handler 的代码（在这种情况下，它只是调用正确的服务来处理该事件）：

public async handle(event: Event): Promise<Component> {
  winston.info('[ComponentHandler:handle]');
  return this.componentService
               .componentStatusChanged(event.header.sender, event.header.senderType, event.payload);
}

这是我的 componentService 类的代码（在这种特定情况下，它只是使用负责将事件保存在数据库中的正确存储库）

public async componentStatusChanged(componentId: string, componentType: ComponentType, componentStatus: ComponentStatus): Promise<any> {
  return this.componentRepository
                   .componentStatusChanged(componentId, componentType, componentStatus);
}

这是我的componentRepository的代码：

public async componentStatusChanged(componentId: string, componentType: ComponentType, componentStatus: ComponentStatus): Promise<Component> {
  winston.info('[ComponentRepository::componentStatusChanged] Component %s newStatus %s', componentId, componentStatus);

  try {
    await this.fetch({ name: componentId });
    return this.update(componentId, { status: componentStatus });
  } catch (ex) {
    // Component not exists
    const newComponent: Component = {
                name: componentId,
                type: componentType,
                status: componentStatus,
                updatedAt: new Date().getTime()
     };
     return this.newComponent(newComponent);
  }
}

public async update(componentId: string, newValues: any): Promise<Component> {
  winston.info('[ComponentRepository::update] Updating component %s. New Values: %j', componentId, newValues);
  await this.component.get().update({ name: componentId }, newValues);
  return this.fetch({ name: componentId });
}

public async newComponent(component: Component): Promise<Component> {
  winston.info('[ComponentRepository::newComponent] Creating Component %j', component);
  return this.component.get().create(component);
}

public async fetch(filter: any): Promise<Component> {
  winston.info('[ComponentRepository::fetch] component %j', filter);
  return this.component.get().findOne(filter);
}

问题是第二个服务以正确的顺序接收事件，但它以错误的顺序将它们转发到 gui，遵循一些日志：

第二个组件接收到的事件负载：已启动 --> 分析 --> 正在听 --> 正在分析 --> 正在听 --> 正在分析 --> 正在听.....

从日志中可以看到，console.log('0__...) 打印的事件顺序与console.log('1___....) 打印的事件顺序不同

info: [newEvent] New Event received: STATUS
0___ STARTED
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus STARTED
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"STARTED"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ ANALYZING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [newEvent] New Event received: STATUS
0___ LISTENING
info: [ComponentHandler:handle]
info: [ComponentRepository::componentStatusChanged] Component auth.srv newStatus LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"ANALYZING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [ComponentRepository::update] Updating component auth.srv. New Values: {"status":"LISTENING"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'ANALYZING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS ANALYZING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'ANALYZING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'ANALYZING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS ANALYZING
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
info: [ComponentRepository::fetch] component {"name":"auth.srv"}
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING
1____ { _id: 5af470ae8278f6320b3a00f3,
  name: 'auth.srv',
  type: 'AUTH_SERVICE',
  status: 'LISTENING',
  updatedAt: 1525969070608,
  __v: 0 }
info: [WebSocketService::publishNewEvent] type: STATUS LISTENING'

编辑

当我使用 WebSockets 时，这似乎是某种竞争条件。我尝试单独测试我的组件，没有问题。一旦我使用 websocket 并快速重复地发布事件，回调堆栈上就会发生一些事情。 我该怎么做才能以正确的顺序处理事件？问题似乎出在我尝试在 mongodb 上执行 I/O 操作时。 谢谢

【问题讨论】：

    标签： node.js mongodb typescript websocket


    【解决方案1】：

    我解决了创建单独处理的事件队列的问题。

    如果您有更好的解决方案，请在此处发布。

    constructor() {
 setInterval(async () => {
   if (this.newEvents.length > 0) {
     await this.handleNewEvents();
   }
 }, 50);
}

this.socketClient.on('NEW_EVENT', (event: MyEvent) => {
   this.newEvents.push(event);
});


private async handleNewEvents(): Promise<any> {
  const event = this.newEvents.shift();
  await this.handleEvent(event);
  if (this.newEvents.length > 0) {
     await this.handleNewEvents();
  }
}

private async handleEvent(event: Event): Promise<any> {
   return new Promise((resolve, reject) => {
       for (const handler of this.handlers) {
           if (handler.isApplicable(event.type)) {
                handler.handle(event).then((handledEvent) => {
                  this.webSockerService.publishNewEvent(event.type, handledEvent);
                      resolve();
                  });
                  return;
               }
            }
        winston.error('Handler not found for %s ', event.type);
           reject();
        });
    }
}

    【讨论】：

      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2013-03-15
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多
      热门标签
      Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode