批处理器

批处理器可用于聚合条目（日志/任何数据）并进行批处理。 当 batch_max_size 设置为零时，处理器将立即执行每个条目。将批处理的最大值设置为大于 1 将开始聚合条目，直到达到最大值或超时。

配置

创建批处理器的唯一必需参数是函数。当批处理达到最大值或缓冲区持续时间超过时，函数将被执行。

名称必选项描述
name可选的标识批处理者的唯一标识符
batch_max_size可选的每批的最大大小，默认为 1000
inactive_timeout可选的如果不活动，将刷新缓冲区的最大时间（以秒为单位），默认值为 5
buffer_duration可选的必须先处理批次中最旧条目的最大期限（以秒为单位），默认是 5
max_retry_count可选的从处理管道中移除之前的最大重试次数；默认为 0
retry_delay可选的如果执行失败，应该延迟进程执行的秒数；默认为 1

以下代码显示了如何在你的插件中使用批处理器：

  1. local bp_manager_mod = require("apisix.utils.batch-processor-manager")
  2. ...
  5. local plugin_name = "xxx-logger"
  6. local batch_processor_manager = bp_manager_mod.new(plugin_name)
  7. local schema = {...}
  8. local _M = {
  9.     ...
  10.     name = plugin_name,
  11.     schema = batch_processor_manager:wrap_schema(schema),
  12. }
  15. ...
  20. function _M.log(conf, ctx)
  21.     local entry = {...} -- data to log
  24.     if batch_processor_manager:add_entry(conf, entry) then
  25.         return
  26.     end
  27.     -- create a new processor if not found
  30.     -- entries is an array table of entry, which can be processed in batch
  31.     local func = function(entries)
  32.         -- serialize to json array core.json.encode(entries)
  33.         -- process/send data
  34.         return true
  35.         -- return false, err_msg if failed
  36.     end
  37.     batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
  38. end

批处理器的配置将通过该插件的配置设置。 举个例子：

  1. curl http://127.0.0.1:9080/apisix/admin/routes/1 -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '
  2. {
  3.       "plugins": {
  4.             "http-logger": {
  5.                 "uri": "http://mockbin.org/bin/:ID",
  6.                 "batch_max_size": 10,
  7.                 "max_retry_count": 1
  8.             }
  9.        },
  10.       "upstream": {
  11.            "type": "roundrobin",
  12.            "nodes": {
  13.                "127.0.0.1:1980": 1
  14.            }
  15.       },
  16.       "uri": "/hello"
  17. }'

如果你的插件只使用一个全局的批处理器， 你可以直接使用它：

  1. local entry = {...} -- data to log
  2. if log_buffer then
  3.     log_buffer:push(entry)
  4.     return
  5. end
  8. local config_bat = {
  9.     name = config.name,
  10.     retry_delay = config.retry_delay,
  11.     ...
  12. }
  15. local err
  16. -- entries is an array table of entry, which can be processed in batch
  17. local func = function(entries)
  18.     ...
  19.     return true
  20.     -- return false, err_msg if failed
  21. end
  22. log_buffer, err = batch_processor:new(func, config_bat)
  25. if not log_buffer then
  26.     core.log.warn("error when creating the batch processor: ", err)
  27.     return
  28. end
  31. log_buffer:push(entry)

注意：请确保批处理的最大值（条目数）在函数执行的范围内。 刷新批处理的计时器基于 inactive_timeout 配置运行。因此，为了获得最佳使用效果， 保持 inactive_timeout 小于 buffer_duration。