标签: task-queue

从任务队列Python API概述:

如果您的任务执行接近30秒限制,App Engine将引发您可能捕获的异常,然后快速保存您的工作或日志过程.

那个例外是什么？

我有一个Google App Engine应用程序,可定期处理内存密集型长时间运行任务.我在线程安全模式下在Python2.7运行时使用taskqueue API,因此我的每个实例都同时处理多个任务.结果,我经常遇到这些错误:

在为8个请求提供服务后,超出了137.496 MB的软私有内存限制

处理此请求后,发现处理此请求的进程使用了​​太多内存并被终止.这可能会导致新进程用于您的应用程序的下一个请求.如果经常看到此消息,则可能是应用程序中存在内存泄漏.

据我所知,每个实例每个都承担8个任务,并最终达到软内存限制.这些任务从使用非常少量的内存开始,但最终增长到大约15-20MB.

有没有办法限制告诉App Engine为实例分配不超过5个请求？或者告诉App Engine该任务预计在10分钟内使用20MB内存并进行相应调整？

我不想使用后端API,因为我希望处理任务的实例数量能够自动扩展,但如果这是唯一的方法,我想知道如何构建它.

我正在使用Google App Engine(Python)实时处理一些事件消息.总之,我有100多个任务,我需要在消息进入时快速运行.我尝试了一些方法(延迟库,线程),我认为最好的解决方案是使用任务队列并异步将这些任务添加到队列中我想要.这是我正在做的一个例子.

tasks = []
task = Task(url=url_for('main.endpoints_worker'),params={'id': id})
tasks.append(task.add_async(queue_name='event-message'))

for task in tasks:
    task.get_result()

当我这样做时,我的大部分时间都花在将这些任务添加到队列中.有没有办法加快速度？有更好的方法吗？

说实话,每次运行时都会遇到很多不同的时间.有时我大约100ms(这可能会很好),但有时我大约1s.

我本以为扩展工作会更快,但批量添加到任务队列执行.以下是我所看到的建议方法:

tasks = [Task(url=url_for('main.endpoints_worker'),params={'id': id}) for id in id_list]
rpc = Queue('event-message').add_async(tasks)
rpc.get_result()

更新:由于添加到队列时100任务限制,我需要再次检查此问题. 通过批量创建我的任务(100个组),我已经大大提高了代码的吞吐量,但我仍然不明白为什么将多组任务添加到队列这么快就会减慢速度.一个任务queue.add_async运行<40毫秒没问题.当我做2个或更多queue.add_async时,那个时间变慢了.我很想知道为什么？另外我怎么解决这个问题？

当我在没有异步的情况下添加批量任务时,每个任务都需要<40ms.为什么在使用异步时它们需要更长的时间？

另一个更新我认为问题可能与争用相关,但即使我将这些任务中的每一个添加到不同的队列,我也会得到相同的结果.

我的理解是在每个宏任务之后处理完整的微任务任务队列。

如果是这种情况，为什么在以下代码片段中setTimeout的Promise微任务之后执行回调JavaScript？

console.log('start');

setTimeout(() => {
  console.log("setTimeout");
});

Promise.resolve().then(function() {
  console.log('promise');
});

console.log('end');

这将输出以下内容：

> "start"
> "end"
> "promise"
> "setTimeout"

是因为4ms现代浏览器强加了 ~延迟吗？

来自MDN：

在现代浏览器中，当由于回调嵌套（其中嵌套级别至少为一定深度）或在一定数量的连续间隔后触发连续调用时，setTimeout()/setInterval()调用被限制为至少每 4 毫秒一次。

尽管这表明它仅适用于连续回调嵌套。

根据Bob 叔叔的 Clean Architecture，企业和应用程序业务规则（概念上由命令组成）位于外部接口层以下的层中。因此，无论何时调用接口，接口都可以自由地运行业务规则。但是，在某些情况下，还需要在后台流程中运行业务规则，无论是单独运行还是作为工作流的一部分，由外部接口的输入或作业调度程序触发。在那种情况下，任务队列服务在 Clean Architecture 中的什么位置？

我想在python中构建一个优先级队列,其中队列包含不同的字典及其优先级编号.因此,当调用"get function"时,具有最高优先级(最低编号)的字典将被拉出队列,并且当调用"add function"时,新字典将被添加到队列中并基于其排序优先号码.

请帮忙...

提前致谢!

我有一个共享内存池,许多不同的线程可以从中请求分配.从这个请求分配将在每个线程中发生很多,但是线程的数量可能很小,通常只有1个线程在运行.我不确定以下哪种方法可以解决这个问题.

最终我可能需要实现两者并看看哪个产生更有利的结果......我也担心即使考虑#2也可能是过早的优化,因为我实际上并没有使用这个共享资源编写的代码.但问题是如此有趣,以至于它继续分散我对其他工作的注意力.

1)创建互斥锁并让线程在获取分配之前尝试锁定它,然后解锁它.

2)让每个线程注册一个请求槽,当需要分配时,它将请求放入槽中,然后阻塞(while(result == NULL){usleep()})等待请求槽有结果.单个线程连续迭代请求时隙,进行分配并将它们分配给请求时隙中的结果.

数字1是简单的解决方案,但如果时机正确,单个线程可能会占用锁定.第二个更复杂,但是当从资源中提取时,确保线程之间的公平性.但是它仍然会阻塞请求线程,如果有很多线程,迭代可以在没有进行任何实际分配的情况下刻录周期,直到找到要满足的请求.

注意:使用pthreads在Linux上使用C语言

我很难理解Google AppEngine Push Task Queue for java的某些方面,即执行任务时,响应在哪里？

如果我像这样添加一些东西到队列:

Queue queue = QueueFactory.getDefaultQueue();
queue.add(withUrl("/worker").param("key", key));

然后在某个时候执行任务,如下所示:

public class SomeServlet extends HttpServlet {
  @Override
  protected void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException, IOException {
    ...
    resp.getWriter().println("something");  //-- where does this response go???
  }
}

对不起新手问题,但这个回复在哪里？我的应用程序没有设置正常的http请求,所以它不等待正常的http响应.我必须遗漏任务队列的基本内容.

我正在使用GAE推送任务队列.我希望任务运行一次,如果失败,我希望它以某种方式通知我.我也不希望它进行任何重试.如果它失败了,我希望它在队列中,直到我手动重新运行它.

我可以手动捕获所有异常并向自己发送电子邮件,但它失败了.有没有更好的办法？在队列配置中,我可以指定零重试限制(见下文),但如果我这样做,任务一旦失败就会被删除.

# queue.yaml
- name: default
  rate: 5/s
  retry_parameters:
    task_retry_limit: 0

有任何想法吗？

我几乎完成了将我的iOS应用程序迁移到Swift 3.0.但我仍然有一些类似于下面的情况.其中大部分我都能通过将有问题的代码放在主线程上来解决.

在其他一些情况下,我无法弄清楚,我的代码的哪一部分正在错误的线程上执行.我收到这样的消息:

This application is modifying the autolayout engine from a background thread after the engine was accessed from the main thread. This can lead to engine corruption and weird crashes.
 Stack:(
    0   CoreFoundation                      0x000000018765a1d8 <redacted> + 148
    1   libobjc.A.dylib                     0x000000018609455c objc_exception_throw + 56
    2   CoreFoundation                      0x000000018765a108 <redacted> + 0
    3   Foundation                          0x0000000188241ea4 <redacted> + 192
    ....................
    16  libsystem_pthread.dylib             0x00000001866eece4 <redacted> + 200
    17  libsystem_pthread.dylib             0x00000001866ee378 pthread_mutex_lock + 0
    18  libsystem_pthread.dylib             0x00000001866edda4 start_wqthread + 4
)

是否有一些特殊的技术(使用调试器时的选项或??)我可以用来跟踪progran所遵循的路径,看看这发生了什么？

setTimeout下面是同步代码之后执行的例子

console.log('hello');
setTimeout(() => console.log('timeout'), 0);
console.log('end');

控制台输出：hello end timeout

异步：

function asyncForEach(array, callBack) {
    array.forEach(x => setTimeout(callBack(x), 0));
}
asyncForEach([1,2,3,4], (i) => console.log(i)); 
console.log('end');

控制台输出：

1
2
3
4
end

同步：

[1,2,3,4].forEach(x => console.log(x));
console.log('end');

控制台输出：

1
2
3
4
end

我试图更好地理解event loop和task queue，并且在观看了一个很好的推荐视频（https://www.youtube.com/watch?v=8aGhZQkoFbQ）后，遇到了以下半声明：“回调可以是同步和异步的”。演讲者继续演示同步方式如何最终不会通过task queue，因此不会通过event loop，因此所有内容都保留在stack. 但是，异步方式会导致task queue用 的最终返回填充setTimeout callback，这将使任何同步的下面代码可执行，因为event loop必须等待一个空stack来插入返回的callback.

运行上面的代码时（我从他那里偷来的东西也编辑过，所以这可能是我的错误添加），异步方式产生与同步方式相同的结果。

有人可以帮助解释为什么异步方式不作为第一个setTimeout示例和/或提供解释如何不将常规回调（例如数组帮助程序方法）插入task …

 File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/core/handlers/exception.py", line 55, in inner
    response = get_response(request)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/core/handlers/base.py", line 197, in _get_response
    response = wrapped_callback(request, *callback_args, **callback_kwargs)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/views/generic/base.py", line 103, in view
    return self.dispatch(request, *args, **kwargs)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/django/views/generic/base.py", line 142, in dispatch
    return handler(request, *args, **kwargs)
  File "/workspace/apps/account/views.py", line 235, in post
    taskqueue.add(queue_name='create-waybill', url = reverse('_api:createWaybill') ,  params = {'key':url_key})

  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/google/appengine/api/taskqueue/taskqueue.py", line 575, in create_rpc
    return apiproxy_stub_map.UserRPC('taskqueue', deadline, callback)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/google/appengine/api/apiproxy_stub_map.py", line 444, in __init__
    self.__rpc = CreateRPC(service, stubmap)
  File "/layers/google.python.pip/pip/lib/python3.8/site-packages/google/appengine/api/apiproxy_stub_map.py", line 69, in CreateRPC …