如何批量调用含有阻塞操作的函数, 并等待全部返回才继续执行下面的逻辑

[zhangshiqian1214]

#1175

给一个可以直接用的

local skynet = require "skynet"

---@class parallels
local _M = {}

local _MT = {__index = _M}

---@return parallels
local function new()
    return setmetatable(
        {
            list = {},
            boot_co = nil,
            boot_error = nil
        },
        _MT
    )
end

local function wakeup_waitco(self, err)
    if not self.boot_error then
        self.boot_error = err
    end
    local boot_co = self.boot_co
    if boot_co then
        self.boot_co = nil
        skynet.wakeup(boot_co)
    end
end

local function wait_end(self)
    assert(not self.boot_co, "allready in wait %s", tostring(self.boot_co))
    self.boot_co = coroutine.running()
    if not next(self.list) then
        skynet.yield()
    else
        skynet.wait(self.boot_co)
    end
    if self.boot_error then
        error(self.boot_error)
    end
end

---@param self parallels
---@param func fun(...):...
---@param ... any
---@return void
function _M.add(self, func, ...)
    local token = {}
    local list = self.list
    list[token] = true
    skynet.fork(
        function(...)
            token.co = coroutine.running()
            local ok, err = xpcall(func, debug.traceback, ...)
            if not ok then
                wakeup_waitco(self, err)
            else
                list[token] = nil
                if not next(list) then
                    wakeup_waitco(self)
                end
            end
        end,
        ...
    )
end

---@param self parallels
---@return void
function _M.wait(self)
    return wait_end(self)
end

return new

使用

local pa = require("parallels")()
local ret1,ret2
pa:add(function()
    ret1 = skynet.call(service1, ...)
end)
pa:add(function()
    ret2 = skynet.call(service2, ...)
end)

pa:wait()

print(ret1,ret2)

在某种情况下, 比如批量调含有http操作的函数里面, 因为不是直接调用skynet.call, 所以skynet.select不适用这种情况, 对parallels进行测试, 发现有时会出现还没有等所有操作完成, 就唤醒了wait

[sue602]

看一下 test/testselect.lua 是否能满足你需求

[zhangshiqian1214]

你把HTTP和数据放到独立的服务，再CALL不是一样吗？

上面我只是这样举例, 实际上会更复杂一些, 不方便把这些逻辑分到独立的服务

[sniper00]

你把HTTP和数据放到独立的服务，再CALL不是一样吗？

上面我只是这样举例, 实际上会更复杂一些, 不方便把这些逻辑分到独立的服务

你可以仿照这个写一个skynet版 https://github.com/sniper00/moon/blob/master/lualib/moon/task.lua

[zhangshiqian1214]

你把HTTP和数据放到独立的服务，再CALL不是一样吗？

上面我只是这样举例, 实际上会更复杂一些, 不方便把这些逻辑分到独立的服务

你可以仿照这个写一个skynet版 https://github.com/sniper00/moon/blob/master/lualib/moon/task.lua

老兄的moon开源项目写的挺漂亮的, star一个
按之前云大的思路改了一版, 虽然达到目的, 不过还是没select的好用

local class = require "base.class"
local skynet = require "skynet"

local Parallel = class()
function Parallel:ctor(total)
   self.total = total
   self.count = 0
end


function Parallel:_wakeup()
   self.count = self.count + 1
   if self.count >= self.total then
       skynet.wakeup(self)
   end
end

function Parallel:add(func, ...)
   skynet.fork(function(f, ...)
       local ok, err = xpcall(f, debug.traceback, ...)
       if not ok then
           skynet.error("parallel run err:", err)
       end
       self:_wakeup()
   end, func, ...)
end

function Parallel:wait()
   skynet.wait(self)
end

return Parallel

[cloudwu]

你把HTTP和数据放到独立的服务，再CALL不是一样吗？

上面我只是这样举例, 实际上会更复杂一些, 不方便把这些逻辑分到独立的服务

不方便分到独立的服务，你还可以 call 自己的。比如你可以增加一个新的服务通道叫 "invoke" 和 "lua" 并列，专门处理自己 call 自己的请求。

这样最终绕回了同一个服务。

[zhangshiqian1214]

你把HTTP和数据放到独立的服务，再CALL不是一样吗？

上面我只是这样举例, 实际上会更复杂一些, 不方便把这些逻辑分到独立的服务

不方便分到独立的服务，你还可以 call 自己的。比如你可以增加一个新的服务通道叫 "invoke" 和 "lua" 并列，专门处理自己 call 自己的请求。

这样最终绕回了同一个服务。

我明白了, 是可以这么做

[cloudwu]

调用一个本地的函数可以转换为 skynet.call ， 我随手写了一段代码，没测试：

skynet.dispatch("invoke", function(session, address, id)
	local f = task[id]
	task[id] = nil
	skynet.ret(skynet.pack(f())
end)


local task = {}

local function invoke(func)
	local n = #task+1
	task[n] = func
	return skynet.self(), "invoke", n
end


--skynet.call(invoke(function()
	--- do something
--	end))

[zhangshiqian1214]

已测试, 没有问题

[cloudfreexiao]

skynet.select 支持 cluster 相关调用嘛

[tengfe]

感谢，飘雪已经收到已经了。Thanks, snow has been received already