Повторно используемая схема сообщений транзакций на основе RabbitMQ

задняя часть
Повторно используемая схема сообщений транзакций на основе RabbitMQ

помещение

Распределенная транзакция представляет собой сложную проблему в практике микросервисов.В схеме практики микросервисов, реализованной автором, принят компромисс или избегание строгой согласованности. Ссылаться наEbayСхема локальной таблицы сообщений, предложенная много лет назад на основеRabbitMQа такжеMySQL(JDBC) обеспечивает облегченную инкапсуляцию и реализует малоинвазивный модуль сообщений о транзакциях. Содержание этой статьи заключается в подробном анализе дизайнерских идей и реализации всей схемы. Зависимости среды следующие:

  • JDK1.8+
  • spring-boot-start-web:2.x.x,spring-boot-start-jdbc:2.x.x,spring-boot-start-amqp:2.x.x
  • HikariCP:3.x.x(spring-boot-start-jdbcсамостоятельный),mysql-connector-java:5.1.48
  • redisson:3.12.1

Идеи оформления схемы

В принципе, сообщения о транзакциях подходят только для слабой согласованности (или"возможная согласованность"), общие сценарии слабой согласованности, такие как:

  • Пользовательская служба завершает действие регистрации и отправляет маркетинговое сообщение в службу SMS.
  • В кредитной системе после того, как служба заказов сохраняет заказ, она отправляет запись об ожидающем заказе в службу утверждения.
  • ......

"Сообщения о транзакциях, как правило, не следует использовать в сценариях со строгой согласованностью.".

Как правило, для указания строгой синхронизации требуется строгая согласованность, то есть все операции должны завершаться успешно или завершаться ошибкой одновременно, что приведет к дополнительному потреблению, вызванному синхронизацией. Если модуль сообщений о транзакциях правильно спроектирован и такие функции, как компенсация, запрос, мониторинг и т. д., выполнены, поскольку взаимодействие системы является асинхронным, общая пропускная способность выше, чем у строгой синхронизации. В бизнес-системе, за которую отвечает автор, базовый принцип также настраивается на основе использования сообщений о транзакциях:"Исходя из того, что содержание сообщения правильное, потребитель должен позаботиться о себе, если есть исключение.".

❝

Проще говоря: восходящий поток обеспечивает правильность своего собственного бизнеса, и когда правильное сообщение успешно отправлено в RabbitMQ, обязательство восходящего потока считается выполненным.

❞

Чтобы уменьшить инвазивность кода, сообщения о транзакциях должны использоватьSpringиз"Программные транзакции"или"декларативная сделка". Программные транзакции обычно полагаются наTransactionTemplate, в то время как декларативные транзакции полагаются наAOPмодули, в зависимости от аннотаций@Transactional.

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

  • Сохраните журнал сообщений.
  • отправить сообщение наRabbitMQСервер.
  • Запрос записей сообщений, отправка компенсации и т. д.

Логическая единица выполнения транзакции

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

отправить сообщениеRabbitMQЭтот шаг на стороне сервера необходимо отложить до"После фиксации транзакции", чтобы убедиться, что транзакция успешно зафиксирована и сообщение успешно отправленоRabbitMQДве операции на стороне сервера одинаковы. чтобы положить"Сохранить ожидающие сообщения о транзакциях"а также"Отправить сообщение RabbitMQ"Два действия объединены в одно действие с точки зрения восприятия пользователем, которое необходимо использовать здесь.SpringСобственный синхронизатор транзакцийTransactionSynchronization, вот анализ callback позиции основного метода синхронизатора транзакций, основной справочникAbstractPlatformTransactionManager#commit()илиAbstractPlatformTransactionManager#processCommit()метод:

На приведенном выше рисунке показан только сценарий, в котором транзакция зафиксирована правильно (исключая сценарий исключения). Здесь видно, что синхронизатор транзакцийTransactionSynchronizationизafterCommit()а такжеafterCompletion(int status)методы находятся в реальной точке фиксации транзакцииAbstractPlatformTransactionManager#doCommit()Вызывается позже, поэтому один из этих двух методов может быть использован для выполнения push-сообщения дляRabbitMQНа стороне сервера общий псевдокод выглядит следующим образом:

@Transactional
public Dto businessMethod(){
    business transaction code block ...
    // 保存事务消息
    [saveTransactionMessageRecord()]
    // 注册事务同步器 - 在afterCommit()方法中推送消息到RabbitMQ
    [register TransactionSynchronization,send message in method afterCommit()]
    business transaction code block ...
}

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

Компенсация за транзакционные сообщения

Хотя я упоминал ранее, что автор предложил нижестоящему сервису позаботиться об аномальном потреблении своего собственного сервиса, иногда вышестоящему необходимо повторно запушить соответствующее сообщение из-за беспомощности Это особая сцена. Есть еще один сценарий для рассмотрения: синхронизатор транзакций срабатывает после фиксации транзакции.TransactionSynchronizationизafterCommit()метод не удался. Это маловероятный сценарий, но он обязательно произойдет в продакшене. Типичная причина:"После того, как транзакция зафиксирована, она еще не успела сработатьTransactionSynchronization#afterCommit()метод для отправки экземпляра службы перезапускается". Как показано ниже:

Чтобы единообразно решить проблему отправки компенсации, используется конечное состояние, чтобы определить, успешно ли было отправлено сообщение:

  • В методе транзакции при сохранении сообщения транзакции отметьте статус отправки записи сообщения как"Обработка".
  • Интерфейс синхронизатора транзакцийTransactionSynchronizationизafterCommit()В реализации метода поместите соответствующее сообщение вRabbitMQ, затем измените статус записи сообщения транзакции на"Нажмите успешно".

Другой очень частный случайRabbitMQСбой самого сервера и отправка сообщения ненормальная, в этом случае требуется повторная попытка (компенсационная отправка)."Опыт показал, что повторные попытки в течение короткого промежутка времени бессмысленны.", неисправный сервис, как правило, не восстанавливается мгновенно, поэтому вы можете рассмотреть возможность использования"Экспоненциальный алгоритм отсрочки"Выполняется повторная попытка, и максимальное количество повторных попыток должно быть ограничено.

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

Реализация программы

Ввести основные зависимости:

<properties>
    <spring.boot.version>2.2.4.RELEASE</spring.boot.version>
    <redisson.version>3.12.1</redisson.version>
    <mysql.connector.version>5.1.48</mysql.connector.version>
</properties>
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>${spring.boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>${mysql.connector.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>${redisson.version}</version>
    </dependency>
</dependencies>

spring-boot-starter-jdbc,mysql-connector-javaа такжеspring-boot-starter-aopдаMySQLсвязанные с транзакциями иspring-boot-starter-amqpдаRabbitMQинкапсуляция клиента,redissonОн в основном использует свои распределенные блокировки для компенсации блокировки выполнения задач по времени (для предотвращения одновременного выполнения принудительной компенсации несколькими узлами в службе).

дизайн стола

Модуль сообщений о транзакциях в основном состоит из двух таблиц,MySQLНапример, создать таблицуDDLследующим образом:

CREATE TABLE `t_transactional_message`
(
    id                  BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    create_time         DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
    edit_time           DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    creator             VARCHAR(20)     NOT NULL DEFAULT 'admin',
    editor              VARCHAR(20)     NOT NULL DEFAULT 'admin',
    deleted             TINYINT         NOT NULL DEFAULT 0,
    current_retry_times TINYINT         NOT NULL DEFAULT 0 COMMENT '当前重试次数',
    max_retry_times     TINYINT         NOT NULL DEFAULT 5 COMMENT '最大重试次数',
    queue_name          VARCHAR(255)    NOT NULL COMMENT '队列名',
    exchange_name       VARCHAR(255)    NOT NULL COMMENT '交换器名',
    exchange_type       VARCHAR(8)      NOT NULL COMMENT '交换类型',
    routing_key         VARCHAR(255) COMMENT '路由键',
    business_module     VARCHAR(32)     NOT NULL COMMENT '业务模块',
    business_key        VARCHAR(255)    NOT NULL COMMENT '业务键',
    next_schedule_time  DATETIME        NOT NULL COMMENT '下一次调度时间',
    message_status      TINYINT         NOT NULL DEFAULT 0 COMMENT '消息状态',
    init_backoff        BIGINT UNSIGNED NOT NULL DEFAULT 10 COMMENT '退避初始化值,单位为秒',
    backoff_factor      TINYINT         NOT NULL DEFAULT 2 COMMENT '退避因子(也就是指数)',
    INDEX idx_queue_name (queue_name),
    INDEX idx_create_time (create_time),
    INDEX idx_next_schedule_time (next_schedule_time),
    INDEX idx_business_key (business_key)
) COMMENT '事务消息表';

CREATE TABLE `t_transactional_message_content`
(
    id         BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    message_id BIGINT UNSIGNED NOT NULL COMMENT '事务消息记录ID',
    content    TEXT COMMENT '消息内容'
) COMMENT '事务消息内容表';

Поскольку этот модуль может быть расширен до модуля фонового управления, поля сообщений, связанные с управлением и состоянием, а также содержимое сообщений большого объема должны храниться в двух таблицах соответственно, чтобы избежать необходимости запрашивать записи сообщений в больших пакетах.MySQLСлужитьIOПроблема высокого использования (это то же самое, что и у предыдущей компании).DBAБолее разумное решение было получено после обсуждения командой). Два бизнес-поля зарезервированыbusiness_moduleа такжеbusiness_keyИспользуется для идентификации бизнес-модулей и бизнес-ключей (обычно это уникальные идентификационные номера, такие как номера заказов).

В общем, если сервис декларирует отношения привязки между очередью и обменом заранее через конфигурацию, то отправляетRabbitMQКогда новости на самом деле зависят только отexchangeNameа такжеroutingKeyдва поля (headerТип обмена особый и редко используется, поэтому здесь пока не рассматривается), учитывая, что сервис может опустить операцию объявления, при отправке сообщения первое объявление привязки будет сделано на основе очереди и соответствующая информация будет кэшироваться (RabbitMQВ объявлении привязки очереди-обмена, если параметры отношения привязки каждый раз согласованы, исключение не будет выдано).

Дизайн кода схемы

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

Определение класса сущностей модели анемииTransactionalMessageа такжеTransactionalMessageContent:

@Data
public class TransactionalMessage {

    private Long id;
    private LocalDateTime createTime;
    private LocalDateTime editTime;
    private String creator;
    private String editor;
    private Integer deleted;
    private Integer currentRetryTimes;
    private Integer maxRetryTimes;
    private String queueName;
    private String exchangeName;
    private String exchangeType;
    private String routingKey;
    private String businessModule;
    private String businessKey;
    private LocalDateTime nextScheduleTime;
    private Integer messageStatus;
    private Long initBackoff;
    private Integer backoffFactor;
}

@Data
public class TransactionalMessageContent {

    private Long id;
    private Long messageId;
    private String content;
}

затем определитеdaoИнтерфейс (подробный код реализации здесь пока раскрываться не будет, использование хранилищаMySQL, если вы хотите заменить базу данных другого типа, вам просто нужно использовать другую реализацию):

public interface TransactionalMessageDao {

    void insertSelective(TransactionalMessage record);

    void updateStatusSelective(TransactionalMessage record);

    List<TransactionalMessage> queryPendingCompensationRecords(LocalDateTime minScheduleTime,
                                                               LocalDateTime maxScheduleTime,
                                                               int limit);
}

public interface TransactionalMessageContentDao {

    void insert(TransactionalMessageContent record);

    List<TransactionalMessageContent> queryByMessageIds(String messageIds);
}

Затем определите интерфейс службы сообщений о транзакциях.TransactionalMessageService:

// 对外提供的服务类接口
public interface TransactionalMessageService {

    void sendTransactionalMessage(Destination destination, TxMessage message);
}


@Getter
@RequiredArgsConstructor
public enum ExchangeType {

    FANOUT("fanout"),

    DIRECT("direct"),

    TOPIC("topic"),

    DEFAULT(""),

    ;

    private final String type;
}

// 发送消息的目的地
public interface Destination {

    ExchangeType exchangeType();

    String queueName();

    String exchangeName();

    String routingKey();
}

@Builder
public class DefaultDestination implements Destination {

    private ExchangeType exchangeType;
    private String queueName;
    private String exchangeName;
    private String routingKey;

    @Override
    public ExchangeType exchangeType() {
        return exchangeType;
    }

    @Override
    public String queueName() {
        return queueName;
    }

    @Override
    public String exchangeName() {
        return exchangeName;
    }

    @Override
    public String routingKey() {
        return routingKey;
    }
}

// 事务消息
public interface TxMessage {

    String businessModule();

    String businessKey();

    String content();
}

@Builder
public class DefaultTxMessage implements TxMessage {

    private String businessModule;
    private String businessKey;
    private String content;

    @Override
    public String businessModule() {
        return businessModule;
    }

    @Override
    public String businessKey() {
        return businessKey;
    }

    @Override
    public String content() {
        return content;
    }
}

// 消息状态
@RequiredArgsConstructor
public enum TxMessageStatus {

    /**
     * 成功
     */
    SUCCESS(1),

    /**
     * 待处理
     */
    PENDING(0),

    /**
     * 处理失败
     */
    FAIL(-1),

    ;

    private final Integer status;
}

TransactionalMessageServiceКласс реализации представляет собой реализацию основной функции сообщений транзакций. Код выглядит следующим образом:

@Slf4j
@Service
@RequiredArgsConstructor
public class RabbitTransactionalMessageService implements TransactionalMessageService {

    private final AmqpAdmin amqpAdmin;
    private final TransactionalMessageManagementService managementService;

    private static final ConcurrentMap<String, Boolean> QUEUE_ALREADY_DECLARE = new ConcurrentHashMap<>();

    @Override
    public void sendTransactionalMessage(Destination destination, TxMessage message) {
        String queueName = destination.queueName();
        String exchangeName = destination.exchangeName();
        String routingKey = destination.routingKey();
        ExchangeType exchangeType = destination.exchangeType();
        // 原子性的预声明
        QUEUE_ALREADY_DECLARE.computeIfAbsent(queueName, k -> {
            Queue queue = new Queue(queueName);
            amqpAdmin.declareQueue(queue);
            Exchange exchange = new CustomExchange(exchangeName, exchangeType.getType());
            amqpAdmin.declareExchange(exchange);
            Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs();
            amqpAdmin.declareBinding(binding);
            return true;
        });
        TransactionalMessage record = new TransactionalMessage();
        record.setQueueName(queueName);
        record.setExchangeName(exchangeName);
        record.setExchangeType(exchangeType.getType());
        record.setRoutingKey(routingKey);
        record.setBusinessModule(message.businessModule());
        record.setBusinessKey(message.businessKey());
        String content = message.content();
        // 保存事务消息记录
        managementService.saveTransactionalMessageRecord(record, content);
        // 注册事务同步器
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                managementService.sendMessageSync(record, content);
            }
        });
    }
}

Управление статусом записи сообщения и сохранением содержимого объединено вTransactionalMessageManagementServiceсередина:

@Slf4j
@RequiredArgsConstructor
@Service
public class TransactionalMessageManagementService {

    private final TransactionalMessageDao messageDao;
    private final TransactionalMessageContentDao contentDao;
    private final RabbitTemplate rabbitTemplate;

    private static final LocalDateTime END = LocalDateTime.of(2999, 1, 1, 0, 0, 0);
    private static final long DEFAULT_INIT_BACKOFF = 10L;
    private static final int DEFAULT_BACKOFF_FACTOR = 2;
    private static final int DEFAULT_MAX_RETRY_TIMES = 5;
    private static final int LIMIT = 100;

    public void saveTransactionalMessageRecord(TransactionalMessage record, String content) {
        record.setMessageStatus(TxMessageStatus.PENDING.getStatus());
        record.setNextScheduleTime(calculateNextScheduleTime(LocalDateTime.now(), DEFAULT_INIT_BACKOFF,
                DEFAULT_BACKOFF_FACTOR, 0));
        record.setCurrentRetryTimes(0);
        record.setInitBackoff(DEFAULT_INIT_BACKOFF);
        record.setBackoffFactor(DEFAULT_BACKOFF_FACTOR);
        record.setMaxRetryTimes(DEFAULT_MAX_RETRY_TIMES);
        messageDao.insertSelective(record);
        TransactionalMessageContent messageContent = new TransactionalMessageContent();
        messageContent.setContent(content);
        messageContent.setMessageId(record.getId());
        contentDao.insert(messageContent);
    }

    public void sendMessageSync(TransactionalMessage record, String content) {
        try {
            rabbitTemplate.convertAndSend(record.getExchangeName(), record.getRoutingKey(), content);
            if (log.isDebugEnabled()) {
                log.debug("发送消息成功,目标队列:{},消息内容:{}", record.getQueueName(), content);
            }
            // 标记成功
            markSuccess(record);
        } catch (Exception e) {
            // 标记失败
            markFail(record, e);
        }
    }

    private void markSuccess(TransactionalMessage record) {
        // 标记下一次执行时间为最大值
        record.setNextScheduleTime(END);
        record.setCurrentRetryTimes(record.getCurrentRetryTimes().compareTo(record.getMaxRetryTimes()) >= 0 ?
                record.getMaxRetryTimes() : record.getCurrentRetryTimes() + 1);
        record.setMessageStatus(TxMessageStatus.SUCCESS.getStatus());
        record.setEditTime(LocalDateTime.now());
        messageDao.updateStatusSelective(record);
    }

    private void markFail(TransactionalMessage record, Exception e) {
        log.error("发送消息失败,目标队列:{}", record.getQueueName(), e);
        record.setCurrentRetryTimes(record.getCurrentRetryTimes().compareTo(record.getMaxRetryTimes()) >= 0 ?
                record.getMaxRetryTimes() : record.getCurrentRetryTimes() + 1);
        // 计算下一次的执行时间
        LocalDateTime nextScheduleTime = calculateNextScheduleTime(
                record.getNextScheduleTime(),
                record.getInitBackoff(),
                record.getBackoffFactor(),
                record.getCurrentRetryTimes()
        );
        record.setNextScheduleTime(nextScheduleTime);
        record.setMessageStatus(TxMessageStatus.FAIL.getStatus());
        record.setEditTime(LocalDateTime.now());
        messageDao.updateStatusSelective(record);
    }

    /**
     * 计算下一次执行时间
     *
     * @param base          基础时间
     * @param initBackoff   退避基准值
     * @param backoffFactor 退避指数
     * @param round         轮数
     * @return LocalDateTime
     */
    private LocalDateTime calculateNextScheduleTime(LocalDateTime base,
                                                    long initBackoff,
                                                    long backoffFactor,
                                                    long round) {
        double delta = initBackoff * Math.pow(backoffFactor, round);
        return base.plusSeconds((long) delta);
    }

    /**
     * 推送补偿 - 里面的参数应该根据实际场景定制
     */
    public void processPendingCompensationRecords() {
        // 时间的右值为当前时间减去退避初始值，这里预防把刚保存的消息也推送了
        LocalDateTime max = LocalDateTime.now().plusSeconds(-DEFAULT_INIT_BACKOFF);
        // 时间的左值为右值减去1小时
        LocalDateTime min = max.plusHours(-1);
        Map<Long, TransactionalMessage> collect = messageDao.queryPendingCompensationRecords(min, max, LIMIT)
                .stream()
                .collect(Collectors.toMap(TransactionalMessage::getId, x -> x));
        if (!collect.isEmpty()) {
            StringJoiner joiner = new StringJoiner(",", "(", ")");
            collect.keySet().forEach(x -> joiner.add(x.toString()));
            contentDao.queryByMessageIds(joiner.toString())
                    .forEach(item -> {
                        TransactionalMessage message = collect.get(item.getMessageId());
                        sendMessageSync(message, item.getContent());
                    });
        }
    }
}

Здесь есть небольшая оптимизация: метод обновления состояния записи сообщения транзакции может быть оптимизирован как пакетное обновление, вlimitЧем он больше, тем выше эффективность пакетного обновления.

Наконец, класс конфигурации для запланированной задачи:

@Slf4j
@RequiredArgsConstructor
@Configuration
@EnableScheduling
public class ScheduleJobAutoConfiguration {

    private final TransactionalMessageManagementService managementService;

    /**
     * 这里用的是本地的Redis,实际上要做成配置
     */
    private final RedissonClient redisson = Redisson.create();

    @Scheduled(fixedDelay = 10000)
    public void transactionalMessageCompensationTask() throws Exception {
        RLock lock = redisson.getLock("transactionalMessageCompensationTask");
        // 等待时间5秒,预期300秒执行完毕,这两个值需要按照实际场景定制
        boolean tryLock = lock.tryLock(5, 300, TimeUnit.SECONDS);
        if (tryLock) {
            try {
                long start = System.currentTimeMillis();
                log.info("开始执行事务消息推送补偿定时任务...");
                managementService.processPendingCompensationRecords();
                long end = System.currentTimeMillis();
                long delta = end - start;
                // 以防锁过早释放
                if (delta < 5000) {
                    Thread.sleep(5000 - delta);
                }
                log.info("执行事务消息推送补偿定时任务完毕,耗时:{} ms...", end - start);
            } finally {
                lock.unlock();
            }
        }
    }
}

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

Наконец добавьте два тестовых класса:

@RequiredArgsConstructor
@Component
public class MockBusinessRunner implements CommandLineRunner {

    private final MockBusinessService mockBusinessService;

    @Override
    public void run(String... args) throws Exception {
        mockBusinessService.saveOrder();
    }
}

@Slf4j
@RequiredArgsConstructor
@Service
public class MockBusinessService {

    private final JdbcTemplate jdbcTemplate;
    private final TransactionalMessageService transactionalMessageService;
    private final ObjectMapper objectMapper;

    @Transactional(rollbackFor = Exception.class)
    public void saveOrder() throws Exception {
        String orderId = UUID.randomUUID().toString();
        BigDecimal amount = BigDecimal.valueOf(100L);
        Map<String, Object> message = new HashMap<>();
        message.put("orderId", orderId);
        message.put("amount", amount);
        jdbcTemplate.update("INSERT INTO t_order(order_id,amount) VALUES (?,?)", p -> {
            p.setString(1, orderId);
            p.setBigDecimal(2, amount);
        });
        String content = objectMapper.writeValueAsString(message);
        transactionalMessageService.sendTransactionalMessage(
                DefaultDestination.builder()
                        .exchangeName("tm.test.exchange")
                        .queueName("tm.test.queue")
                        .routingKey("tm.test.key")
                        .exchangeType(ExchangeType.DIRECT)
                        .build(),
                DefaultTxMessage.builder()
                        .businessKey(orderId)
                        .businessModule("SAVE_ORDER")
                        .content(content)
                        .build()
        );
        log.info("保存订单:{}成功...", orderId);
    }
}

Результат теста следующий:

2020-02-05 21:10:13.287  INFO 49556 --- [           main] club.throwable.cm.MockBusinessService    : 保存订单:07a75323-460b-42cb-aa63-1a0a45ce19bf成功...

Данные смоделированного заказа успешно сохранены, иRabbitMQСообщение обычно отправляется после того, как транзакция успешно зафиксированаRabbitMQсервер, напримерRabbitMQпоказаны консольные данные.

резюме

Конструкция модуля транзакционных сообщений предназначена только для того, чтобы реализация функции отправки асинхронных сообщений была завершенной.На самом деле, разумная система взаимодействия с асинхронными сообщениями определенно обеспечивает синхронный интерфейс запроса, который основан на функции, которую асинхронные сообщения нет обратного звонка или нет ответа. Вообще говоря, пропускная способность системы положительно связана с долей асинхронной обработки системы (по этому вопросу см.Amdahl's Law), поэтому при проектировании фактической архитектуры системы следует максимально использовать асинхронное взаимодействие, чтобы повысить пропускную способность системы и сократить ненужное ожидание, вызванное синхронной блокировкой. Модуль сообщений о транзакциях может быть расширен до фонового управления и может даже взаимодействовать сMicrometer,Prometheusа такжеGrafanaСистема осуществляет мониторинг данных в режиме реального времени.

эта статьяdemoРепозиторий проекта:rabbit-transactional-message

demoДолжен быть установлен локальноMySQL,Redisа такжеRabbitMQДля нормального запуска новую базу данных необходимо создать локально.local.

(Конец этой статьи c-5-d e-a-20200202 Эпидемия серьезная, скоро я начну работать из дома, и я буду меньше читать и больше читать)

Технический публичный аккаунт Throwable Digest (id: throwable-doge) время от времени публикует оригинальные технические статьи автора (никогда не занимайтесь плагиатом и не перепечатывайте):

В этой статье используетсяmdniceнабор текста

Категории