Фоновые задачи в App Engine

GAE - очень полезный инструмент для создания масштабируемых веб-приложений. Некоторые из ограничений, на которые указывают многие, включают отсутствие поддержки фоновых задач, отсутствие периодических задач и строгие ограничения на то, сколько времени занимает каждый HTTP-запрос; если запрос превышает этот временной лимит, операция завершается, что делает невозможным выполнение трудоемких задач. .

Как запустить фоновую задачу?
В GAE код выполняется только при наличии HTTP-запроса. Существует строгий лимит времени (я думаю, 10 секунд) на то, сколько времени может занять код. Поэтому, если запросов нет, код не выполняется. Одним из предложенных способов решения проблемы было использование внешнего ящика для непрерывной отправки запросов, чтобы создать фоновую задачу. Но для этого нам нужен внешний бокс, и теперь мы зависим от еще одного элемента. Другой альтернативой была отправка ответа перенаправления 302, чтобы клиент повторно отправил запрос, это также делает нас зависимыми от внешнего элемента, который является клиентом. Что, если этот внешний блок - это сама GAE? Каждый, кто использовал функциональный язык, который не поддерживает конструкцию цикла, знает об альтернативе, т.е. рекурсия является заменой цикла. Так что, если мы завершим часть вычислений и выполним HTTP GET на том же URL-адресе с очень коротким тайм-аутом, скажем, 1 секунда? Это создает цикл (рекурсию) в PHP-коде, запущенном на apache.

<?php
$i = 0;
if (isset($_REQUEST["i"])){
        $i= $_REQUEST["i"];
    sleep(1);
}
$ch = curl_init("http://localhost".$_SERVER["PHP_SELF"]."?i = ".($i+1));
curl_setopt($ch, CURLOPT_HEADER, 0);
curl_setopt($ch, CURLOPT_TIMEOUT, 1);
curl_exec($ch);
print "hello world\n";
?>

Кое-как это не работает на GAE. Так что, если мы выполним HTTP GET на каком-то другом URL-адресе, например url2, который выполняет HTTP GET на первом URL-адресе? Кажется, это работает в GAE. Код для этого выглядит так.

class FirstUrl(webapp.RequestHandler):
    def get(self):
        self.response.out.write("ok")
        time.sleep(2)
        urlfetch.fetch("http://"+self.request.headers["HOST"]+'/url2')

class SecondUrl(webapp.RequestHandler):
    def get(self):
        self.response.out.write("ok")
        time.sleep(2)
        urlfetch.fetch("http://"+self.request.headers["HOST"]+'/url1')

application = webapp.WSGIApplication([('/url1', FirstUrl), ('/url2', SecondUrl)])
def main():
    run_wsgi_app(application)
if __name__ == "__main__":
    main()

Поскольку мы обнаружили способ запуска фоновой задачи, давайте создадим абстракции для периодической задачи (таймер) и конструкцию цикла, которая охватывает множество HTTP-запросов (foreach).

Таймер
Теперь строить таймер проще простого. Основная идея состоит в том, чтобы иметь список таймеров и интервал, с которым каждый из них должен вызываться. Как только мы достигнем этого интервала, вызовите функцию обратного вызова. Мы будем использовать кэш памяти для ведения списка таймеров. Чтобы узнать, когда вызывать обратный вызов, мы сохраним ключ в кэше памяти с интервалом в качестве времени истечения срока действия. Мы периодически (скажем, 5 секунд) проверяем, присутствует ли этот ключ, если нет, то вызываем обратный вызов и снова устанавливаем этот ключ с интервалом.

def timer(func, interval):
    timerlist = memcache.get('timer')
    if (None == timerlist):
        timerlist = []
    timerlist.append({'func':func, 'interval':interval})
    memcache.set('timer-'+func, '1', interval)
    memcache.set('timer', timerlist)

def checktimers():
    timerlist = memcache.get('timer')
    if (None == timerlist):
        return False
    for current in timerlist:
        if (None == memcache.get('timer-'+current['func'])):
            #reset interval
            memcache.set('timer-'+current['func'], '1', current['interval'])
            #invoke callback function
            try:
                eval(current['func']+'()')
            except:
                pass
            return True
    return False

Для каждого
Это необходимо, когда мы хотим выполнить длительные вычисления, скажем, выполнить некоторую операцию с 1000 строками базы данных или получить 1000 URL-адресов и т. д. Основная идея состоит в том, чтобы поддерживать список обратных вызовов и аргументов в кэше памяти и каждый раз вызывать обратный вызов с аргументом.

def foreach(func, args):
    looplist = memcache.get('foreach')
    if (None == looplist):
        looplist = []
    looplist.append({'func':func, 'args':args})
    memcache.set('foreach', looplist)

def checkloops():
    looplist = memcache.get('foreach')
    if (None == looplist):
        return False
    if ((len(looplist) > 0) and (len(looplist[0]['args']) > 0)):
        arg = looplist[0]['args'].pop(0)
        func = looplist[0]['func']
        if (len(looplist[0]['args']) == 0):
            looplist.pop(0)
        if ((len(looplist) > 0) and (len(looplist[0]['args']) > 0)):
            memcache.set('foreach', looplist)
        else:
            memcache.delete('foreach')
        try:
            eval(func+'('+repr(arg)+')')
        except:
            pass
        return True
    else:
        return False

# instead of
# foreach index in range(0, 1000):
#   someoperaton(index)
# we will say
# foreach('someoperaton', range(0, 1000))

Теперь можно легко создать программу, которая получает список URL-адресов каждый час. Вот код.

def getone(url):
    try:
        result = urlfetch.fetch(url)
        if (result.status_code == 200):
            memcache.set(url, '1', 60*60)
            #process result.content
    except :
        pass

def getallurl():
    #list of urls to be fetched
    urllist = ['http://www.google.com/', 'http://www.cnn.com/', 'http://www.yahoo.com', 'http://news.google.com']
    fetchlist = []
    for url in urllist:
        if (memcache.get(url) is None):
            fetchlist.append(url)
    #this is equivalent to
    #for url in fetchlist: getone(url)
    if (len(fetchlist) > 0):
        foreach('getone', fetchlist)

#register the timer callback
timer('getallurl', 3*60)

полный код здесь http://groups.google.com/group/httpmr-discuss/t/1648611a54c01aa Я запускал этот код на appengine в течение нескольких дней без особых проблем.

Предупреждение: мы активно используем urlfetch. Предел количества urlfetch в день составляет 160000. Так что будьте осторожны, чтобы не достигнуть этого лимита.

В грядущей версии среды выполнения будет какой-то механизм периодического выполнения a'la cron. См. это сообщение в группе AppEngine.

So, all the SDK pieces appear to work, but my testing indicates this isn't running on the production servers yet-- I set up an "every 1 minutes" cron that logs when it runs, and it hasn't been called yet

Но когда это будет доступно, сказать сложно ...

Вы можете использовать API очереди задач Python.

Используйте очередь задач - http://code.google.com/appengine/docs/java/taskqueue/overview.html

Если вы хотите запускать периодические фоновые задачи, см. этот вопрос (AppEngine cron)

Если ваши задачи не периодические, см. API очереди задач Python или API очереди задач Java

Дополнительные сведения о заданиях cron можно найти в Python App Engine здесь.

В движок приложения встроено средство cron.

Пожалуйста, обратитесь к: https://developers.google.com/appengine/docs/python/config/cron?hl=en

Использование Отложенная библиотека Python - это самый простой способ выполнить фоновую задачу в Appengine с использованием Python, который построен поверх TaskQueue API.

from google.appengine.ext import deferred

def do_something_expensive(a, b, c=None):
    logging.info("Doing something expensive!")
    # Do your work here

# Somewhere else
deferred.defer(do_something_expensive, "Hello, world!", 42, c=True)