Node.js中Shopify产品批量插入的实践优化与疑问

我通过Shopify API监听新产品创建事件，接收webhook时会拿到如下JSON数据（示例）：

{
    "admin_graphql_api_id": "gid://shopify/Product/8867427221826",
    "body_html": "<strong>Thshirt in verschiedenen farben</strong>",
    "created_at": "2023-10-14T10:51:06-04:00",
    "handle": "t-shirt",
    "id": 8867427221826,
    "product_type": "",
    "published_at": "2023-10-14T10:51:06-04:00",
    "template_suffix": "",
    "title": "T-shirt",
    "updated_at": "2023-10-14T10:51:08-04:00",
    "vendor": "crystelorbs",
    "status": "active",
    "published_scope": "global",
    "tags": "",
    "variants": [
        {
            "admin_graphql_api_id": "gid://shopify/ProductVariant/47150807122242",
            "barcode": "",
            "compare_at_price": null,
            "created_at": "2023-10-14T10:51:06-04:00",
            "fulfillment_service": "manual",
            "id": 47150807122242,
            "inventory_management": "shopify",
            "inventory_policy": "deny",
            "position": 1,
            "price": "29.95",
            "product_id": 8867427221826,
            "sku": "321dsfase1231321w",
            "taxable": true,
            "title": "34 / Blau",
            "updated_at": "2023-10-14T10:51:06-04:00",
            "option1": "34",
            "option2": "Blau",
            "option3": null,
            "grams": 0,
            "image_id": null,
            "weight": 0,
            "weight_unit": "kg",
            "inventory_item_id": 49189935579458,
            "inventory_quantity": 0,
            "old_inventory_quantity": 0,
            "requires_shipping": true
        },
        {
            "admin_graphql_api_id": "gid://shopify/ProductVariant/47150807187778",
            "barcode": "",
            "compare_at_price": null,
            "created_at": "2023-10-14T10:51:07-04:00",
            "fulfillment_service": "manual",
            "id": 47150807187778,
            "inventory_management": "shopify",
            "inventory_policy": "deny",
            "position": 2,
            "price": "29.95",
            "product_id": 8867427221826,
            "sku": "321dsfase1231321w-2",
            "taxable": true,
            "title": "33 / Blau",
            "updated_at": "2023-10-14T10:51:07-04:00",
            "option1": "33",
            "option2": "Blau",
            "option3": null,
            "grams": 0,
            "image_id": null,
            "weight": 0,
            "weight_unit": "kg",
            "inventory_item_id": 49189935644994,
            "inventory_quantity": 0,
            "old_inventory_quantity": 0,
            "requires_shipping": true
        },
        {
            "admin_graphql_api_id": "gid://shopify/ProductVariant/47150807220546",
            "barcode": "",
            "compare_at_price": null,
            "created_at": "2023-10-14T10:51:07-04:00",
            "fulfillment_service": "manual",
            "id": 47150807220546,
            "inventory_management": "shopify",
            "inventory_policy": "deny",
            "position": 3,
            "price": "29.95",
            "product_id": 8867427221826,
            "sku": "321dsfase1231321w-3",
            "taxable": true,
            "title": "12 / Blau",
            "updated_at": "2023-10-14T10:51:07-04:00",
            "option1": "12",
            "option2": "Blau",
            "option3": null,
            "grams": 0,
            "image_id": null,
            "weight": 0,
            "weight_unit": "kg",
            "inventory_item_id": 49189935677762,
            "inventory_quantity": 0,
            "old_inventory_quantity": 0,
            "requires_shipping": true
        }
    ],
    "options": [
        {
            "name": "Größe",
            "id": 11148870222146,
            "product_id": 8867427221826,
            "position": 1,
            "values": [
                "34",
                "33",
                "12"
            ]
        },
        {
            "name": "Farbe",
            "id": 11148870254914,
            "product_id": 8867427221826,
            "position": 2,
            "values": [
                "Blau"
            ]
        }
    ],
    "images": [
        {
            "id": 43049062433090,
            "position": 1,
            "product_id": 8867427221826,
            "width": 650,
            "height": 300,
            "alt": null,
            "src": "https://cdn.shopify.com/s/files/1/0757/6479/3666/files/ggg-modified_6dcaf2fa-d307-4ee7-bdf1-c7a9dbeefd62.png?v=1697295067",
            "created_at": "2023-10-14T10:51:07-04:00",
            "updated_at": "2023-10-14T10:51:07-04:00",
            "admin_graphql_api_id": "gid://shopify/ProductImage/43049062433090",
            "variant_ids": []
        }
    ],
    "image": {
        "id": 43049062433090,
        "position": 1,
        "product_id": 8867427221826,
        "width": 650,
        "height": 300,
        "alt": null,
        "src": "https://cdn.shopify.com/s/files/1/0757/6479/3666/files/ggg-modified_6dcaf2fa-d307-4ee7-bdf1-c7a9dbeefd62.png?v=1697295067",
        "created_at": "2023-10-14T10:51:07-04:00",
        "updated_at": "2023-10-14T10:51:07-04:00",
        "admin_graphql_api_id": "gid://shopify/ProductImage/43049062433090",
        "variant_ids": []
    }
}

我用事务创建产品，比如含50张图片的产品要循环执行50次client.query('INSERT INTO ....')。因为Node.js是单线程，这种遍历数组插入图片的方式是不是不良实践？我了解过worker threads，但已经在用队列系统，还需要用worker threads吗？

当前代码如下：

export const CreateProduct = async ({product, shop_id}: { product: _Product; shop_id: UUID; }): Promise<ReturnQuery> => {
  const client = await pool.connect();
  try {
    let product_options_data: {
      p_o_id: number;
      values: string[];
    }[] = [];

    await client.query('BEGIN');

    const Product: QueryResult = await client.query(`
      INSERT INTO product
      (info, name, status, shop_id)
      VALUES
      ($1, $2, $3, $4);
    `, [product.body_html, product.title, product.status, shop_id]);

    const p_id = Product.rows[0].id;

    if(product?.images) {
      for(let i = 0; i < product?.images?.length; i++) {
        await client.query('INSERT INTO product_media (p_id, src, alt, s_position) VALUES ($1, $2, $3, $4);', [p_id, product.images[i].src, product.images[i].alt, product.images[i].position]);
      }
    }

    if(product?.options) {
      if(product.options.length > 0) {
        for(let a = 0; a < product.options.length; a++) {
          if(product?.options[a].values) {
            const p_o_id: QueryResult = await client.query('INSERT INTO product_options (p_id, name, position) VALUES ($1, $2, $3);', [p_id, product.options[a].name, product.options[a].position]);
            product_options_data.push({
              p_o_id: p_o_id.rows[0].id,
              values: product.options[a].values as string[]
            })
          }
        }
      }
    }

    await client.query('COMMIT');
  } catch(e) {
    await client.query('ROLLBACK');
    console.log(e);
    Sentry.captureException(e);
    return {
      message: ErrorMessageQuery,
      statusCode: 500,
      query: 'SHOPIFY_CREATE_PRODUCT'
    }
  } finally {
    client.release();
  }
};

我想知道有没有优化这段代码的方法，听说要保持事务简短，但产品有大量图片或变体时必须遍历插入所有数据，求优化方案建议。

关于单线程循环插入与Worker Threads的问题 #

单线程下循环执行await client.query确实是低效的——每次插入都要等待数据库响应，50张图片就会产生50次网络往返和数据库处理开销，属于不良实践。但你已经在用队列系统，不需要额外引入Worker Threads：队列已经把webhook请求的处理异步化，不会阻塞主服务的其他请求；而Worker Threads更多用于CPU密集型任务，数据库操作是IO密集型，用队列足够隔离负载。

核心优化方向 #

1. 批量插入替代循环单条插入 #

这是最有效的优化，把多条INSERT合并成一个语句，大幅减少数据库交互次数：

• 图片批量插入（安全参数化写法）：

  if(product?.images && product.images.length > 0) {
    // 构建批量插入的占位符和参数数组
    const placeholders = product.images.map((_, idx) => 
      `($1, $${idx*3+2}, $${idx*3+3}, $${idx*3+4})`
    ).join(',');
    const params = [p_id];
    product.images.forEach(img => {
      params.push(img.src, img.alt || null, img.position);
    });
    await client.query(
      `INSERT INTO product_media (p_id, src, alt, s_position) VALUES ${placeholders};`,
      params
    );
  }
  

• 产品选项批量插入：同理，先收集所有选项的参数，再用批量语法插入，避免多次单独查询的开销。

2. 缩短事务时长 #

事务越长，数据库锁占用时间越久，容易引发并发问题。可以：

• 把非核心、允许延迟的数据操作移出事务（比如图片如果允许后续异步同步，可放到队列里单独处理，不放在主事务中）；
• 确保所有批量操作在事务内一次性完成，避免不必要的等待。

3. 连接池与资源管理优化 #

当前代码中client.release()放在finally块是正确的，但要注意：

• 批量操作时尽量复用同一个连接，不要频繁获取释放；
• 调整连接池的配置（比如最大连接数），避免批量操作时连接耗尽。

4. 异步并行处理（可选） #

如果数据库支持并行查询，可以用Promise.all同时发起多个批量插入请求（注意事务内的并行需要数据库支持，且不要超过连接池限制）：

const promises = [];
if(product?.images && product.images.length > 0) {
  // 准备图片批量插入的promise
  promises.push(/* 图片批量插入的query */);
}
if(product?.options && product.options.length > 0) {
  // 准备选项批量插入的promise
  promises.push(/* 选项批量插入的query */);
}
await Promise.all(promises);

总结 #

• 循环单条插入是不良实践，优先用批量插入优化；
• 已用队列系统的情况下，不需要额外引入Worker Threads；
• 事务尽量保持简短，把可异步的操作移出主事务，或者用批量操作压缩事务内的执行时间；
• 始终使用参数化查询避免SQL注入，不要直接拼接字符串。

内容的提问来源于stack exchange，提问作者asdasd