Выбор сценария и реализация асинхронных компонентов

Автор: Xianyu Technology - Code Treasure

Сводка перспектив

В нашем ежедневном процессе разработки мы часто сталкиваемся с не связанными между собой людьми.流水账逻辑(завершение данных, логика уведомлений и т. д.), в настоящее время мы обычно используем асинхронную обработку для ускорения ответа. В то же время, с увеличением количества восходящих и нисходящих зависимых сервисов, также может быть сгенерирован ряд соответствующих сервисов. , Вопросы включают, но не ограничиваются问题难以排查, 性能难以保证и т. д. в Xianyu не исключение:

• дополненный串行操作(Здесь имеется в виду последовательный интерфейс для достижения нескольких сетевых операций ввода-вывода) Существует узкое место в производительности, которое серьезно влияет на интерфейс rt.
• Степень унификации логики неудовлетворительна, «свободный полет» логики приводит к ухудшению читаемости кода, а показатели, связанные с логикой модуля, трудно измерить.

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

1. Мониторинг: все состояния и поведение (успех, сбой, rt и другие ключевые индикаторы) в рамках логического жизненного цикла каждой единицы находятся в пределах области мониторинга.
2. Устойчивость к катастрофам: существует резервный механизм, который может быть полезен, когда в логической единице возникает большое количество исключений (обычно тайм-аутов).
3. Бесплатный доступ: простой доступ и использование, так что его можно использовать сразу после установки.

взять каштан

сценарий без тайм-аута

Предполагая, что логический блок L имеет три задачи a, b и c, время выполнения равно t1=1, t2=2, t3=3, а время ожидания равно 4. Как показано на рисунке:

Желаемый результат сейчас:

1. Время выполнения для логического блока L 3
2. Время, необходимое для успешного выполнения задачи один раз, равно 1, время для успешного выполнения задачи b равно 2, а время для успешного выполнения задачи c равно 3.

Сценарий тайм-аута

Предположим, что логический блок L имеет три задачи a, b и c, время выполнения которых равно t1=3, t2=5, t3=6, а время ожидания равно 4. Как показано на рисунке:

Желаемый результат сейчас:

1. Время выполнения логического блока L равно 4
2. Задача a требует 3 для успешного выполнения один раз, задача b не выполняется, а задача c не выполняется.
3. B и C Нитки останавливаются после сбоя и не продолжают заниматься резенными ресурсами пула нитей

план

Обратите внимание, что все приведенные ниже решения основаны на超时场景расширять

Схема акка

1. Текущая служба получает соответствующий LogicActor
2. LogicActor получает все текущие UnitActor(a, b, c)
3. использоватьtellСпособ соответственно сообщения a, b, c, обратите внимание, что этот процесс является асинхронным
4. a возвращается нормально, b и c оба тайм-аута
5. LogicActor перейдет к текущей задаче обратного отсчета независимо от времени ожидания
6. Вернитесь к объединению данных, когда все задачи будут выполнены
7. Наконец, верните данные результата

LogicActor — субъект логической единицы, отвечающий за распределение задач и объединение данных.

Unitactor- Специфические задачи актер

преимущество

1. управляемое сообщением.
2. Нет необходимости в дополнительном управлении пулом потоков и устойчивости к исключениям.
3. Изящно остановите выполняющийся рабочий процесс (PoinsonPill).
4. Те, кто знаком с akka, могут легко начать работу.

недостаток

1. Для достижения цели требуется слой инкапсуляции, а это непросто.
2. Значительно увеличить сложность системы.
3. Это катастрофа для тех, кто не знаком со scala/akka.

Стоимость обучения scala / akka напрямую привела к провалу цели «доступа с нулевой стоимостью», поэтому от нее отказались.

Схема rxjava

Логика реализована с использованием rxjava

1. Определите CountDownLatch для управления временем ожидания.
2. Определите три задания
3. Реализовать асинхронизацию с помощью RXJAVA
4. Дайте пул потоков фиксированного размера 10 для обработки
5. Операция накопления в уменьшении
6. Защелка как единая карточная точка

Тайм-аут здесь влияет на общий балл карты, но я не знаю, какое дело вызвало тайм-аут.

преимущество

1. реактивное программирование
2. Уменьшение размера кода
3. Блокировать работу пула потоков
4. в упаковке#timeoutи#onErrorReturnметод, существующий модуль обработки тайм-аута не нужно инкапсулировать дважды

недостаток

1. Для достижения цели требуется слой упаковки
2. Существует модуль обработки времени ожидания, но внедрение бизнес-единицы уничтожит исходный пакет RXJAVA, и сложность преобразования велика.

Хотя "толерантность к катастрофам"(onErrorReturn), но это не может быть «отслеживаемым», потому что после тайм-аута у нас нет возможности узнать, какой бизнес истек, поэтому мы сдаемся.

Комплексное решение - пакет на основе пакета JUC

трудность

Как реализовать мониторинг логического блока

1. Глобально внутри группыtraceIdИнкапсулировать (ThreadLocal), используя пул потоков丢失上下文, так что соответствующий журнал невозможно отследить.
2. существуетbizCodeконцепции удобно контролировать логическую единицу и каждую операцию внутри логической единицы.

Как очистить потоки тайм-аута, чтобы они не занимали ресурсы пула потоков

вышесказанноеCallableиRunnableПо истечении времени выполнения его необходимо остановить, чтобы он не продолжал занимать ресурсы пула потоков.

Как контролировать параллелизм (тайм-аут)

Есть много инструментов для контроля тайм-аута

• CountDownLatch: java.util.concurrent.CountDownLatch#await(long, java.util.concurrent.TimeUnit) Тайм-аут не вызовет исключения. Для определения превышения тайм-аута можно использовать только логическое значение, что означает, что его нельзя обрабатывается потоком исключений короткого замыкания.
• CyclicBarrier: java.util.concurrent.CyclicBarrier#await(long, java.util.concurrent.TimeUnit) не подходит для инкапсуляции исключений, например
  • Теперь есть четыре потока a, b, c, d.Время ожидания составляет 2 с.Из них поток d выполняется в течение 3 с, и происходит следующее
    1. Первая завершенная задача для входа в барьер. Ожидание вызовет исключение тайм-аута, а остальные четыре задачи вызовут сломанный барьер
    2. Задача тайм-аута, наконец, завершит последующие действия и продолжит занимать ресурсы.
• Семафор: Не соответствует условиям использования, но подробнее.
• Future: java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit), futureList будет считываться последовательно, в результате чего T>=Max(t1,t2,t3)

Типичный пример ошибки

1. Отправить будущее в соответствующий список
2. Получите соответствующий результат, перейдя
3. После получения результата проделайте операцию накопления
4. Последний результат вывода и время выполнения

Хотя указанные выше три задачи выполняются одновременно, из-заfuture.get(long, TimeUnit)блокируется, тайм-аут здесь не пройдет. Как показано выше, окончательное время может быть3+2+1=6s.

Общая программа

Диаграмма основных классов выглядит следующим образом:

1. ConcurrentCallable — это класс, используемый извне.
2. BizBaseCallable наследует Callable, который инкапсулирует внутренний контекст и соответствующее бизнес-подразделение.
3. Для решения проблемы мониторинга тайм-аутов бизнеса BizCountDownLatch добавляет атрибуты пула бизнес-единиц
4. BizBaseCallable инкапсулирует контекст представления ctx, чтобы гарантировать, что контекст не будет потерян в случае многопоточности (совместимость с traceId внутри группы)

Модернизация CountDownLatch — возможность мониторинга

Унаследуйте CountDownLatch и переопределите методы #await() и countDown().

1. Концепция дополнительной бизнес-единицы (bizCode)
2. Соответствующий bizSet на самом деле является незавершенным бизнес-пулом.После завершения бизнеса удалите соответствующую бизнес-единицу
3. Переопределите метод countDown, чтобы представить его, удалив бизнес-подразделение.已完成
4. Исключение выдается, когда время ожидания истекает, имитируя среду короткого замыкания, и верхний уровень обрабатывает его единообразно.
5. Незавершенная задача выброшена в исключение тайм-аута

Инкапсуляция логики мониторинга верхнего уровня

Обработка тайм-аута ConcurrentCallable — аварийное восстановление

1. 监控逻辑Содержит обработку логики тайм-аута
2. Когда Callable получает возвращаемое значение, период ожидания получения дается0: Полные завершены, не завершены сверхурочно.
3. Прервите выполнение вызываемого объекта с помощью future.cancel
4. По умолчанию возвращается откат (обратная операция), а откат реализуется по умолчанию какrenturn null;.
5. Здесь необходимо подставить значение по умолчанию, чтобы его можно было легко передать при вызове внешнего слоя.list.forEach(l -> deal(l))для обработки логики списка

Посмотрите на приведенный выше пример — доступ с нулевой стоимостью

1. Размещение контекста моделирования, ThreadLocal инкапсулируется в TestCtx
2. 4000L указывает, что время ожидания составляет 4000 мс, что называется именем «L».
3. Добавьте задачи a, b, c в компонент
4. Наконец, показана обработка после получения данных.

1. После завершения a он будет вcountDownLatchУдалите соответствующий блок из
2. Когда время истечет, b и c будут принудительно завершены

Это завершает可监控, 容灾性, 零成本接入Асинхронный компонент .

резюме

Таким образом, мы наконец выбрали «JUC (обновленная версия)» в качестве нашего асинхронного компонента.

В этой статье в основном представлены компоненты асинхронного планирования, используемые внутри Xianyu.Бизнес-компоненты, полученные из бизнес-сценариев, могут не обладать сильной универсальностью, но я надеюсь дать некоторые полезные уроки из предыстории, целей и выбора схемы в тексте. .