如何使用事件总线实现微服务间通信？传输数据为空如何排查

核心错误点 #

• HTTP方法使用错误：所有服务的/events接口都用了GET方法，GET请求规范中不建议携带请求体，Express的req.body默认不会解析GET请求的body内容，所以拿到的永远是空值。
• 重复响应问题：同一个请求路由中多次调用res.json()/res.send()，会抛出异常，且后调用的响应不会生效。
• 发送端逻辑错误：
  • 执行的是find()查询操作，返回的doc是查询结果数组，不存在insertedCount属性（该属性仅插入操作返回），对应的分支判断永远不会生效
  • 发送事件的axios请求写在响应返回之后，可能因请求上下文销毁导致请求未正常发出
  • axios.get参数格式错误，第二个参数是请求配置对象，直接传自定义数据不会被当成请求体发送
• 事件总线逻辑错误：转发事件时axios.get参数错误，同样存在重复响应问题
• 接收端逻辑错误：用GET接口读取req.body，永远返回空

修复代码示例 #

1. 发送端代码修正 #

// 所有/events相关请求全换成POST方法
app.get ('/products', async (req , res) => {
    try {
        let db = await connect();
        let cursor = await db.collection('posts').find({});
        let doc = await cursor.toArray();

        // 先发送事件，再返回响应
        await axios.post('http://localhost:4205/events', {
            type: 'Success',
            data: {
                doc,
                postId: req.params.id, // 注意：如果/products路由没有配置id参数，需自行调整该字段
            }
        })

        // 单次返回响应，不要重复调用res.send/res.json
        res.json(doc);
        
        // 如果你原逻辑是插入操作，把上面的find替换为insert，再用insertedCount判断即可
        /* 插入逻辑参考
        let result = await db.collection('posts').insertOne(待插入数据)
        if (result.insertedCount == 1) {
            await axios.post('http://localhost:4205/events', {
                type: 'Success',
                data: {
                    _id: result.insertedId,
                    postId: req.params.id,
                }
            })
            res.send({
                status: 'success',
                id: result.insertedId,
            });
        } else {
            res.send({
                status: 'fail',
            });
        }
        */
    } catch (e) {
        res.status(500).send({status: 'error', msg: e.message})
    }
});

2. 事件总线代码修正 #

// 换成POST方法
app.post('/events', async (req, res) => {
    try {
        const event = req.body;
        // 转发事件到下游服务
        await axios.post('http://localhost:4202/events', event)
        // 单次返回响应
        res.send({status:'OK'})
    } catch (e) {
        res.status(500).send({status: 'error', msg: e.message})
    }
})

3. 接收端微服务代码修正 #

// 换成POST方法
app.post('/events', async (req, res) => {
    // 此处可正常拿到事件数据
    console.log('收到事件：', req.body)
    res.send(req.body)
});

额外注意事项 #

• 确保所有服务都配置了Express的body解析中间件，否则req.body还是会为空：

const express = require('express')
const app = express()
// 加入这两行解析json和表单格式的请求体
app.use(express.json())
app.use(express.urlencoded({extended: true}))

• 所有异步操作比如axios请求、数据库操作都要加await或者catch捕获异常，避免静默失败无法定位问题。

内容的提问来源于stack exchange，提问作者Jan Tuđan