CQRS в Windows Azure

Microsoft Windows Azure предлагает уникальные возможности (эластичную масштабируемость, сокращение затрат и гибкость развертывания) и в то же время ставит перед разработчиками уникальные задачи, потому что среда Windows Azure отличается от стандартных Windows-серверов, на которых сегодня размещена большая часть сервисов и приложений Microsoft .NET Framework.

Один из самых веских доводов в пользу размещения приложений и сервисов в облаке — эластичная масштабируемость (elastic scalability): вы наращиваете мощности для своего сервиса, только когда в этом возникает необходимость, а потом сокращаете их, если количество обращений к сервису уменьшается. В Windows Azure наиболее эффективный способ регулирования выделяемых мощностей — горизонтальное масштабирование (scaling out) вместо вертикального (scaling up), т. е. добавление большего количества серверов, а не модернизация существующих серверов. Для соответствия этой модели масштабирования приложение должно быть динамически масштабируемым. В данной статье описывается эффективный подход к созданию масштабируемых сервисов и демонстрируется, как реализовать их в Windows Azure.

Command Query Responsibility Segregation (CQRS) — это новый подход к созданию масштабируемых приложений. Применяемая в нем архитектура может выглядеть не такой, как вы привыкли в .NET, но она опирается на проверенные временем принципы и решения, нацеленные на достижение масштабируемости. По тематике написания масштабируемых приложений можно найти колоссальные массивы знаний, но это дело требует некоторых перемен в образе мышления.

В переносном смысле, CQRS — не более чем заявление о разделении обязанностей (separation of concerns), но в контексте программной архитектуры оно зачастую обозначает набор связанных шаблонов. Другими словами, термин CQRS может иметь два значения: шаблон и архитектурный стиль. В этой статье я кратко обрисую обе стороны этого термина, а также приведу примеры на основе веб-приложения, выполняемого в Windows Azure.

Что такое шаблон CQRS

Терминология CQRS уходит своими корнями в язык с объектно-ориентированным шаблоном. Команда — это операция, изменяющая состояние чего-либо, а запрос — операция, извлекающая информацию о состоянии. Поэтому команды фактически являются операциями записи, а запросы — операциями чтения.

Шаблон CQRS просто утверждает, что операции чтения и записи нужно моделировать явным образом как раздельные обязанности. Запись данных — это одна обязанность, а их чтение — другая. В большинстве приложений требуется и то, и другое, но, как показано на рис. 1, каждую обязанность можно обрабатывать раздельно.

Рис. 1. Отделение операций чтения от операций записи

Приложение пишет в систему, концептуально отличную от той, откуда оно читает.

Очевидно, что данные, которые записывает приложение, должны в конечном счете стать доступными для чтения. В шаблоне CQRS ничего не говорится о том, как это должно происходить, но в простейшей реализации системы чтения и записи могли бы использовать одно и то же нижележащее хранилище данных.

Итак, операции чтения и записи строго разделяются; операции записи никогда не возвращают данные. Это обманчиво тривиальное утверждение открывает широкие возможности для создания приложений с высокой масштабируемостью.

Архитектурный стиль CQRS

Концепция архитектурного стиля CQRS проста, но предлагает весьма впечатляющую реализацию отображения данных. Рассмотрим рис. 2, на котором показан UI приложения бронирования; в данном случае допустим, что это система заказа столиков в ресторане.

Рис. 2. Отображаемые данные устаревают в момент их визуализации

Календарь показывает дни в текущем месяце, но некоторые даты неактивны, так как на эти дни все столики уже заказаны.

Насколько актуальны данные в таком UI? За время, которое уходит на передачу данных по сети, их визуализацию, восприятие пользователем и реагирование, информация могла уже измениться в нижележащем хранилище данных. Чем дольше пользователь смотрит на эту информацию, тем больше она устаревает. Кроме того, пользователя может отвлечь телефонный звонок или что-то еще, поэтому время принятия решения может измеряться в минутах.

Распространенный способ решения этой проблемы — использование оптимистичной параллельной обработки в тех случаях, когда происходят конфликты. Разработчики приложений должны писать код, способный обрабатывать такие ситуации, но вместо их интерпретации как исключительных случаев в архитектурном стиле CQRS предлагается считать их базовым условием. Когда показываемые данные устаревают в момент визуализации, они не должны отражать информацию в центральном хранилище данных. Вместо этого приложение может выводить информацию из денормализованного источника данных, состояние которого может немного отставать от состояния «настоящего» хранилища данных.

Понимание того, что отображаемые данные всегда являются устаревшими, в сочетании с принципом CQRS, согласно которому операции записи никогда не возвращают данные, создает условия для масштабирования. В UI не следует ждать записи данных, а просто посылать асинхронное сообщение и возвращать пользователю некое представление. Фоновые рабочие потоки принимают сообщения и обрабатывают их в своем темпе. На рис. 3 приведена более подробная схема архитектуры в стиле CQRS.

Рис. 3. Архитектура в стиле CQRS

Всякий раз, когда приложению нужно обновить данные, оно посылает команду в виде асинхронного сообщения — скорее всего через длительно существующий запрос. Как только команда отправлена, UI может вернуть пользователю представление. Фоновый рабочий поток принимает сообщение-команду в отдельном процессе и записывает соответствующие изменения в хранилище данных. В ходе этой операции также генерируется событие как еще одно асинхронное сообщение. Другие обработчики сообщений могут подписаться на такие события и соответственно обновлять денормализованное представление хранилища данных.

Хотя данные в этом представлении будут отставать от «настоящих», распространение этого события зачастую происходит настолько быстро, что пользователи не успевают что-либо заметить. Но, даже если работа системы замедляется из-за избыточной нагрузки, данные в представлении в конечном счете будут находиться в согласованном состоянии.

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

Приложение для заказа столиков в ресторане

Простое приложение для заказа столиков в ресторане служит отличным примером того, как реализуется CQRS в Windows Azure. Вообразите, что это приложение принимает запросы на заказы столиков в ресторане. На первой странице размещен элемент управления DatePicker, как показано на рис. 2; вновь обратите внимание на то, что некоторые даты недоступны, т. е. все столики на эти дни уже заказаны.

Когда пользователь щелкает доступную дату, отображается форма резервирования, а потом уведомление о приеме заказа (рис. 4).

Рис. 4. Схема работы UI в процессе резервирования

Заметьте, что страница уведомления о приеме заказа информирует пользователя о том, что в данный момент выполнение заказа не гарантируется, но будут приложены все усилия, чтобы его выполнить. Окончательный ответ будет выслан по электронной почте.

UI в CQRS играет важную роль в формировании ожиданий, так как обработка происходит в фоне. Однако при нормальной нагрузке страница уведомления дает достаточный выигрыш по времени, чтобы после ее прочтения запрос уже был обработан и пользователь получил ответ по почте.

Теперь я продемонстрирую ключевые моменты в реализации приложения-примера для резервирования. Так как даже в этом простом приложении много «движущихся деталей», основное внимание будет уделено наиболее интересным фрагментам кода; полный исходный код для этой статьи вы сможете скачать с сайта MSDN Magazine.

Передача команд

Веб-роль реализуется как приложение ASP.NET MVC 2. Когда пользователь отправляет форму, приведенную на рис. 4, вызывается соответствующая операция контроллера (Controller Action):

[HttpPost]
public ViewResult NewBooking(BookingViewModel model)
{
  this.channel.Send(model.MakeNewReservation());
  return this.View("BookingReceipt", model);
}

Поле channel — это встраиваемый экземпляр простого интерфейса IChannel:

public interface IChannel
{
  void Send(object message);
}

Команда, которую метод NewBooking посылает по каналу, представляет собой данные HTML-формы, инкапсулированные в Data Transfer Object (DTO). Метод MakeNewReservation просто преобразует отправленные данные в экземпляр MakeReservationCommand:

public MakeReservationCommand MakeNewReservation()
{
  return new MakeReservationCommand(this.Date,
    this.Name, this.Email, this.Quantity);
}

Поскольку метод Send ничего не возвращает (void), UI может вернуть HTML-страницу пользователю, как только команда будет успешно отправлена. Реализация интерфейса IChannel поверх очереди гарантирует, что метод Send вернет управление максимально быстро.

В Windows Azure интерфейс IChannel можно реализовать поверх встроенных очередей, которые являются частью Windows Azure Storage. Чтобы помещать сообщения в такую длительно существующую очередь (durable queue), реализация должна обеспечивать сериализацию сообщений. Это можно делать самыми разными способами, но, чтобы не усложнять пример, я предпочел задействовать двоичный сериализатор, встроенный в .NET Framework. Однако в производственном приложении вы должны всерьез подумать об альтернативах, так как двоичный сериализатор затрудняет контроль версий. Например, что будет, когда новая версия вашего кода попытается десериализовать двоичный объект (blob), сериализованный прошлой версией кода? Возможные альтернативы включают XML, JSON или Protocol Buffers.

С выбранным набором технологий реализация IChannel.Send весьма проста:

public void Send(object command)
{
  var formatter = new BinaryFormatter();
  using (var s = new MemoryStream())
  {
    formatter.Serialize(s, command);
    var msg = new CloudQueueMessage(s.ToArray());
    this.queue.AddMessage(msg);
  }
}

Метод Send сериализует Command и создает новый CloudQueueMessage из полученного байтового массива. Поле queue является встроенным экземпляром класса CloudQueue из Windows Azure SDK. Получив информацию об адресе и удостоверения, метод AddMessage добавляет сообщение в соответствующую очередь. Обычно это выполняется на удивление быстро, поэтому, когда этот метод возвращает управление, вызвавший код может заняться другой работой. В то же время сообщение теперь находится в очереди и ждет своей обработки.

Обработка команд

Если веб-роли отлично отображают HTML и принимают данные, которые они могут посылать через интерфейс IChannel, то рабочие роли (Worker Roles) принимают и обрабатывают сообщения из очереди — каждая в своем темпе. Эти фоновые рабочие роли не имеют состояний и являются автономными, поэтому, если они не успевают справляться с поступающими сообщениями, вы можете добавить дополнительные экземпляры. Именно это и обеспечивает огромную масштабируемость архитектуры на основе сообщений.

Как уже было продемонстрировано, передача сообщений через очереди Windows Azure осуществляется очень легко. Но их безопасное и согласованное использование посложнее. Каждый Command инкапсулирует намерение изменить состояние приложения, поэтому фоновый рабочий поток должен убедиться, что ни одно сообщение не потеряно и что нижележащие данные изменяются так, чтобы не нарушилась их целостность.

Это можно сделать довольно легко, используя технологию очередей, которая поддерживает распределенные транзакции (например, Microsoft Message Queuing). Очереди Windows Azure не являются транзакционными, но в них заложен собственный набор гарантий. Сообщения не удаляются после чтения, а просто скрываются на некий период. Клиенты должны извлекать сообщение из очереди, выполнять соответствующие операции и на последнем этапе удалять сообщение. Именно так и делает универсальная рабочая роль в приложении-примере для заказа столиков в ресторане; она выполняет метод PollForMessage (рис. 5) в бесконечном цикле.

Рис. 5. Метод PollForMessage

public void PollForMessage(CloudQueue queue)
{
  var message = queue.GetMessage();
  if (message == null)
  {
    Thread.Sleep(500);
    return;
  }

  try
  {
    this.Handle(message);
    queue.DeleteMessage(message);
  }
  catch (Exception e)
  {
    if (e.IsUnsafeToSuppress())
    {
      throw;
    }
    Trace.TraceError(e.ToString());
  }
}

Метод GetMessage может вернуть null, если в данный момент в очереди нет никаких сообщений. В этом случае метод просто ждет 500 мс и возвращает управление, в каковом случае он будет немедленно вызван заново в бесконечном внешнем цикле. Получив сообщение, этот метод обрабатывает его, вызывая метод Handle, в котором и выполняется вся реальная работа. Поэтому если Handle вернет управление без генерации исключения, вы можете безопасно удалить сообщение.

С другой стороны, если при обработке сообщения произойдет исключение, важно подавить его; необработанное исключение приведет к краху экземпляра рабочей роли, и она перестанет извлекать сообщения из очереди.

В производственной реализации нужно предусматривать обработку так называемых подозрительных сообщений (poison messages), но я решил оставить ее за рамками этого примера, чтобы не усложнять его.

Если исключение генерируется в процессе обработки сообщения, это сообщение не удаляется. После некоторого периода оно вновь станет доступным для обработки. Это гарантия, предоставляемая очередями Windows Azure: сообщение можно обработать минимум один раз. Как следствие оно может воспроизводиться несколько раз. Таким образом, все фоновые рабочие потоки должны уметь обрабатывать воспроизведение сообщений (повторения) (replays). Это важно для того, чтобы все длительные операции записи были идемпотентными.

Как сделать операции записи идемпотентными

Каждый метод, обрабатывающий какое-либо сообщение, должен уметь работать с повторениями без компрометации состояния приложения. Хороший пример — обработка MakeReservationCommand. Общая схема обработки сообщения показана на рис. 6.

Рис. 6. Рабочий процесс обработки MakeReservationCommand

Первое, что должно делать приложение, — проверять, достаточен ли резерв у ресторана на запрошенный день; на данный день все столики могут быть уже заказаны или их осталось очень мало. Чтобы выяснить ответ на этот вопрос, приложение отслеживает текущий резерв в хранилище. Делать это можно несколькими способами. Один из вариантов — отслеживать все дни резервирования в базе данных SQL Azure, но поскольку на размер этих баз данных накладывается ограничение, более масштабируемым способом является применение либо Windows Azure Blob Storage (хранилище двоичных объектов), либо Windows Azure Table Storage (хранилище таблиц).

Приложение-пример использует хранилище двоичных объектов для сохранения сериализованного идемпотентного объекта-значения. Его класс Capacity отслеживает принятые заказы на столики, чтобы распознавать повторения сообщений. Чтобы выяснить оставшийся резерв, приложение может загрузить экземпляр Capacity для соответствующей даты и вызвать метод CanReserve с идентификатором резервирования (ID):

public bool CanReserve(int quantity, Guid id)
{
  if (this.IsReplay(id))
  {
    return true;
  }
  return this.remaining >= quantity;
}

private bool IsReplay(Guid id)
{
  return this.acceptedReservations.Contains(id);
}

С каждым MakeReservationCommand сопоставлен свой ID. Чтобы обеспечить идемпотентность, класс Capacity сохраняет каждый принятый ID резервирования; это позволяет распознавать повторения. Реальная бизнес-логика запускается, только если вызов метода не является повторением. Запрошенное число мест сравнивается с оставшимся.

Приложение сериализует и хранит экземпляр Capacity для каждой даты, поэтому для проверки оставшегося резерва оно загружает соответствующий двоичный объект и вызывает CanReserve:

public bool HasCapacity(MakeReservationCommand reservation)
{
  return this.GetCapacityBlob(reservation)
    .DownloadItem()
    .CanReserve(reservation.Quantity, reservation.Id);
}

Если результат равен true, приложение запускает набор операций, сопоставленный с этим результатом, как показано на рис. 6. Первый шаг — уменьшение резерва, для чего вызывается метод Capacity.Reserve, приведенный на рис. 7.

Рис. 7. Метод Capacity.Reserve

public Capacity Reserve(int quantity, Guid id)
{
  if (!this.CanReserve(quantity, id))
  {
    throw new ArgumentOutOfRangeException();
  }

  if (this.IsReplay(id))
  {
    return this;
  }

  return new Capacity(this.Remaining - quantity,
    this.acceptedReservations
      .Concat(new[] { id }).ToArray());
}

Это еще одна идемпотентная операция, которая сначала вызывает методы CanReserve и IsReplay. Если вызов представляет новый запрос на резервирование какого-то количества мест, возвращается новый экземпляр Capacity с соответственно уменьшенным резервом, и его ID добавляется в список принятых идентификаторов.

Класс Capacity представляет Value Object (объект значения), поэтому он должен передаваться обратно в Windows Azure Blob Storage до окончания операции. На рис. 8 показано, как исходный двоичный объект изначально загружается из Windows Azure Blob Storage.

Рис. 8. Уменьшение значения Capacity и его возврат в хранилище

public void Consume(MakeReservationCommand message)
{
  var blob = this.GetCapacityBlob(message);
  var originalCapacity = blob.DownloadItem();

  var newCapacity = originalCapacity.Reserve(
    message.Quantity, message.Id);

  if (!newCapacity.Equals(originalCapacity))
  {
    blob.Upload(newCapacity);
    if (newCapacity.Remaining <= 0)
    {
      var e = new SoldOutEvent(message.Date);
      this.channel.Send(e);
    }
  }
}

Это сериализованный экземпляр Capacity, соответствующий дате, на которую запрошен столик в ресторане. Если резерв изменяется (т. е. повторения не было), новый Capacity загружается обратно в хранилище двоичных объектов.

Что произойдет, если в процессе обработки будет сгенерировано исключение? Одна из причин, по которой это могло случиться, — экземпляр Capacity изменился после того, как был вызван метод CanReserve. Это вполне вероятно в крупномасштабных системах, где множество конкурирующих запросов обрабатывается параллельно. В таких случаях метод Reserve мог бы вызвать исключение из-за нехватки оставшегося резерва. Это нормально и просто означает, что данный запрос на резервирование проиграл при конкуренции с другими запросами. Исключение будет перехвачено обработчиком на рис. 5, но сообщение, поскольку оно никуда не делось, позднее вновь появится в очереди и будет обработано еще раз. Тогда метод CanReserve немедленно вернет false, и запрос будет вежливо отклонен.

Однако на рис. 8 потенциально возможны другие конфликты параллельной обработки. Что будет, когда два фоновых рабочих потока одновременно попытаются обновить резерв на один и тот же день?

Применение оптимистичной параллельной обработки

Метод Consume на рис. 8 загружает двоичный объект Capacity из хранилища и возвращает в него новое значение, если оно изменилось. Многие фоновые рабочие потоки могут делать это параллельно, поэтому приложение должно гарантировать, что одно значение не будет перезаписано другим.

Поскольку Windows Azure Storage основано на REST, для решения таких задач параллельной обработки рекомендуется использовать ETag. При первом создании приложением экземпляра Capacity для данной даты ETag будет равен null, но когда из хранилища загружается существующий двоичный объект, у его ETag будет некое значение, доступное через CloudBlob.Properties.ETag. Когда приложение возвращает экземпляр Capacity в хранилище, оно должно присваивать корректное значение AccessCondition в экземпляре BlobRequestOptions:

options.AccessCondition = etag == null ?
  AccessCondition.IfNoneMatch("*") :
  AccessCondition.IfMatch(etag);

Когда приложение создает новый экземпляр Capacity, ETag содержит null, а в свойстве AccessCondition должно быть значение IfNoneMatch("*"). Это гарантирует генерацию исключения, если такой двоичный объект уже существует. С другой стороны, если текущая операция записи представляет собой обновление, AccessCondition следует установить в IfMatch — тогда исключение генерируется, когда ETag в хранилище двоичных объектов не совпадает с переданным ETag.

Оптимистичная параллельная обработка на основе ETag — важная часть вашего инструментария, но вы должны явным образом включать ее, передавая соответствующий BlobRequestOptions.

Если при уменьшении резерва исключения не было, приложение может перейти к следующему этапу на рис. 6: записи заказанных мест в хранилище таблиц. Этот процесс проходит примерно по тем же принципам, что и уменьшение резерва, поэтому я не стану его описывать. Требуемый код содержится в полном пакете исходного кода, который можно скачать для этой статьи, но основной момент, как и прежде, заключается в том, что операция записи должна быть идемпотентной.

Последний этап в этом рабочем процессе — генерация события, уведомляющего, что резервирование принято. Это делается отправкой еще одного асинхронного сообщения через очередь Windows Azure. Любые другие фоновые рабочие потоки, которых интересует это событие, могут получить его и обработать, например отправить подтверждение пользователю по электронной почте. Кроме того, приложение должно закрыть цикл в UI, обновив хранилище данных представления.

Обновление данных представления

События, происходящие при обработке команды, посылаются как асинхронные сообщения через интерфейс IChannel. В качестве примера метод Consume на рис. 8 генерирует SoldOutEvent, если резерв сократился до нуля. Обработчики сообщений могут подписываться на такие события, чтобы должным образом обновлять данные представления:

public void Consume(SoldOutEvent message)
{
  this.writer.Disable(message.Date);
}

Встраиваемый writer реализует метод Disable, который обновляет массив отключенных дней месяца в хранилище двоичных объектов:

public void Disable(DateTime date)
{
  var viewBlob = this.GetViewBlob(date);
  DateTime[] disabledDates = viewBlob.DownloadItem();
  viewBlob.Upload(disabledDates
    .Union(new[] { date }).ToArray());
}

Эта реализация просто загружает массив отключенных экземпляров DateTime из двоичного хранилища, добавляет в массив новую дату и вновь отправляет его в это хранилище. Поскольку используется метод Union, операция идемпотентна, а метод Upload вновь инкапсулирует логику оптимистичной параллельной обработки на основе ETag.

Запрос данных представления

Теперь UI может напрямую запрашивать данные из представления. Это эффективная операция, так как здесь данные статичны и никаких вычислений не требуется. Например, чтобы обновить элемент управления DatePicker на рис. 2 недоступными датами, посылается AJAX-запрос контроллеру, который возвращает соответствующий массив.

Контроллер может просто обработать запрос следующим образом:

public JsonResult DisabledDays(int year, int month)
{
  var data = this.monthReader.Read(year, month);
  return this.Json(data, JsonRequestBehavior.AllowGet);
}

Встраиваемый reader реализует метод Read, который считывает двоичный объект, записываемый обработчиком SoldOutEvent:

public IEnumerable<string> Read(int year, int month)
{
  DateTime[] disabledDates =
    this.GetViewBlob(year, month).DownloadItem();
  return (from d in disabledDates
    select d.ToString("yyyy.MM.dd"));
}

На этом цикл закрывается. Пользователь просматривает сайт, ориентируясь на данные текущего представления и заполняет форму для передачи данных, которая обрабатывается на основе обмена асинхронными сообщениями. Наконец, данные представления обновляются при генерации событий в ходе рабочего процесса.

Денормализация данных

В заключение отмечу, что большинство приложений гораздо чаще считывают данные, чем записывают, поэтому оптимизация чтения способствует масштабируемости — особенно когда данные можно считывать из таких статических ресурсов, как двоичные объекты. Данные, которые выводятся на экран, всегда являются отсоединенными, а значит, устаревают в момент визуализации. В CQRS эта проблема решается разделением чтения и записи данных. Данные не обязательно должны считываться непосредственно из того же источника, куда они записываются. Вместо этого данные можно асинхронно передавать из хранилища, в которое они записываются, в представления, где за проецирование данных и манипуляции над ними вы платите лишь раз.

Благодаря встроенным очередям и масштабируемым хранилищам денормализованных данных Windows Azure отлично подходит для такой архитектуры. Хотя распределенные транзакции не поддерживаются, очереди гарантируют, что сообщения никогда не теряются, а обслуживаются минимум один раз. Для обработки потенциально возможных повторений все асинхронные операции записи должны быть идемпотентными. Для реализации оптимистичной параллельной обработки денормализованных данных, которые содержатся в хранилищах двоичных объектов и таблиц, нужно использовать ETag. В совокупности эти простые методики обеспечат целостность данных.

В этой статье я лишь кратко обрисовал возможности CQRS. Если вы хотите узнать больше о CQRS, используйте множество ресурсов, имеющихся сейчас в Интернете. Однако хорошей отправной точкой является начальная страница по CQRS на сайте Рината Абдуллина: abdullin.com/cqrs.