如何在nodej中的事件发射器中解决数据丢失问题？微服务

Question

下面的代码仅用于测试目的，如果它有效，那么我想用微服务在实际项目中实现相同的操作。我在一台服务器中有4个微服务。微服务之间的通信是通过兔子。

我的流量：

微服务1（通过RabbitMQ从此处发送数据）=====＆gt;微服务2（从RabbitMQ接收数据）将数据恢复到====＆gt;微服务1。

微服务1有两个文件：

1. emitter.controller.js
2. 接收authcontroller.js

microservice 2现在有一个文件：

1. readecteTsettingsController.js

，这是Flow -＆gt;

emitter.controller.js（m1）将数据发送到rectectSettingsController.js（m2），然后再次接收settingscontroller.js（m2）将相同的数据发送回到已接收到authauthcontroller.js（m1）在emitter.controller.js（M1）中收听，此后我们将发送HTTP响应。

首先，我们将所有响应以“键”为单位，作为每个请求的mongodbobjectid，只是为了唯一性，“值”是每个请求的响应。 令emit_response_list_map = new Map（）;

最后，我们通过地图访问响应并将响应寄回。

我的问题是：

1. 如果我通过 jmeter 达到100个请求，那么所有流程都有效，但是在Expive autheauthcontroller.js（m1）中，有时会发出重复值。在emitter.controller.js（M1）中的侦听器中发生2-3次，并且数据丢失发生。 注意：在侦听器中，所有100个发射数据都已收到，但是在100个objectid中，有100个正在重复，所有这些都不是唯一的。如何在事件发射器中解决此数据丢失问题

？ i.sstatic.net/dmvkd.png“ alt =”在此处输入图像说明“>

我的代码下方

microservice 1： 文件：emitter.controller.js

var common = require('../config/event_emitter');
var commonEmitter = common.commonEmitter;
const { sendMessage } = require('../config/rabbitmq/emit');
const connection = require('../config/rabbitmq/connection');
var mongoose = require('mongoose');
const { parse, stringify } = require('flatted');



let emit_response_list_map = new Map();
exports.emitOnceTesting = async (req, res, next) => {
    let count = mongoose.Types.ObjectId();
    emit_response_list_map.set(count.toString(), res);

    const rabbit_conn = await connection.connect();
    const emit_channel = await rabbit_conn.createChannel();
    let method = "emit_r";

    let dataObj = { method, count };
    let dataPacket = { exchange: 'device_model', key: "emit.key", data: dataObj };
    emit_channel.assertExchange('device_model', 'topic', {
        durable: false
    });

    sendMessage(emit_channel, dataPacket);
    console.log(`============================== [[SEND] => [FROM] auth_ms : emit_controller: ${method} [TO] settings_ms : receivedDeviceController: emit_controller: ${method}] =======================================, ${dataPacket.data.count}`);

    commonEmitter.on('emitter_test', async (data) => {
        try {
            console.log("______________emitter_test_______________", data.index);

            commonEmitter.removeAllListeners('emitter_test');
            await emit_channel.close();
            const res = emit_response_list_map.get(data.index);

            res.status(200).json({ status: true });        
        } catch (err) {
            console.log(err);
        }
    });
}

微服务2： - ＆gt;文件：readectsettingscontroller.js：

 let emitTesting_q1 = await device_channel_sett_ms.assertQueue('emit1test_r', { exclusive: true });
        device_channel_sett_ms.bindQueue(emitTesting_q1.queue, 'device_model', 'emit.key');
        device_channel_sett_ms.consume(emitTesting_q1.queue, async (msg) => {
            try {
                // @ts-ignore
                let data = parse(msg.content);
                console.log('============================== [[RECEIVED] => [HERE] settings_ms : receivedDeviceController: emitTesting [FROM] coin_zone_ms : receivedDeviceController: emitTesting] =======================================', data);            //res.json({ status: true, success: data });
                let dataPacketAck = { exchange: 'device_model', key: 'emit.key.ack', data };

                await sendMessage(device_channel_sett_ms, dataPacketAck);
                console.log('============================== [[SEND-ACK] => [FROM] settings_ms : receivedDeviceController: emitTesting [FROM] coin_zone_ms : receivedDeviceController:emitTesting ] =======================================', data);
            } catch (err) {
                console.log('exception in getGeoFenceSetting : getGeoFenceDetails -----------', err);
            }
        },
            {
                noAck: true
            }
        );

微服务1：文件：recdiveAuthController.js

   let emit_key_ack = await device_channel_auth_ms.assertQueue('', { exclusive: true });
    device_channel_auth_ms.bindQueue(emit_key_ack.queue, 'device_model', 'emit.key.ack');
    device_channel_auth_ms.consume(emit_key_ack.queue, async (msg) => {
        // @ts-ignore
        let data = parse(msg.content);
        console.log("___________________________________ack________________", data.countack);
        if (data.method = "emit_r") {
            commonEmitter.emit('emitter_test', data);
        } else if (data.method = "") {
        } else if (data.method = "") {
        }
    });