# delay-service

**Repository Path**: paracraft/delay-service

## Basic Information

- **Project Name**: delay-service
- **Description**: http://code.kp-para.cn/tatfook/delay-service.git 镜像仓库
- **Primary Language**: JavaScript
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-01-22
- **Last Updated**: 2025-06-23

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 基于消息队列的分布式事务解决方案

本方案适用的场景在于保证数据的最终一致性。

## 好处

-   可以提高接口的响应速度

-   解耦，非核心业务挂掉不影响核心业务的正常服务。

## 应用场景

1. 场景一，数据的全量更新时，数据需要同步至其他存储系统时。例如：更新 project 信息，需要同步至 es 服务上，即使 es 服务挂掉，也不会影响 project 的正常更新，而 delay service 会在 es 服务恢复之后重新执行更新的请求

2. 场景二，服务间调用，网络问题或其他问题导致的调用失败后，可以不影响主业务的成功返回，失败的请求通过重试的方式再次执行。例如：提前生成小程序码时，存储服务调用七牛云的接口可能会失败，失败后 delay service 进行重试。

3. 场景三，数据的增量更新，角色续期、增加知识豆等场景。

## 基本组成

delay service 提供了一系列接口 http://yapi.kp-para.cn/project/172/interface/api

### delay service 有一个外部的生产者。

delay service 通过接收内部的请求，将请求生产至相关的队列

### delay service 有三个消费者队列。

1. 无顺序请求的消费者。 包括角色的续期、增加知识豆、包括服务器直传文件到七牛云等场景是无需顺序的。此消费者消费数据时，如果失败，会将结果记录至数据库的 errorLogs 中，然后会自动进行重试，重试时会将错误的请求发送到重试队列，由重试队列的消费者执行重试的操作，由开发者去排查问题，排查完问题，可以批量将错误重试。

2. 有顺序请求的消费者。当对数据全量更新时，比如对 project 往 es 更新，后者的更新会将前者的更新覆盖，所以这种情景是有顺序的。当有顺序的消息消费失败时，将不会继续消费，此时可以监控消费队列的 offset 来进行报警，再去排查错误

3. 错误重试的消费者。此消费者重试错误的消息，原理跟无顺序消费者大体相似。

> 重试策略是根据重试次数 N 来计算下次重试的时间间隔 T: T = 2^N\*30(秒)

## 使用

外部服务通过调用[send 接口](http://yapi.kp-para.cn/project/172/interface/api/3822)

其中延迟的请求需要实现幂等性的要求，发给 delay service 的请求会侵入一个 headers `x-delay-service-trace-id`，此参数也可以由调用方传过来。

接收方创建`delayServiceTraceIds`表`x-delay-service-trace-id`作为主键，再加一个`createdAt`字段，当有请求时，如果有此 header，判断存不存在，存在即不处理直接返回成功，不存在则处理，再插入此 header，由此实现幂等性的简单途径。

> 如果没有`x-delay-service-trace-id`也应该能够正常处理请求

## 水平拓展

部署多个。

## keepwork 需要调整的地方

**每个服务中可以内部调用的更新操作的接口都应该实现幂等性。**

-   es-service: 资源的更新接口

-   git-marshal: ?

-   goods-system: 执行兑换、新增用户物品

-   accounting: 内部可以调用的更新操作接口

-   lesson: /coreApi/user

-   onlineQuiz: registrate

-   push-manager: 创建消息

-   push-server: kickout、msg

-   ts-storage

## 配置

`config/config.env.js`

```
    config.sequelize = {
        dialect: 'mysql',
        host: 'localhost',
        port: 3306,
        database: 'delay_service',
        username: 'root',
        password: null,
    };

    config.kafka = {
        size: 5, // the number of messages read from Kafka
        connection: {
            // 'metadata.broker.list': '10.28.18.7:19092',
            'metadata.broker.list': 'localhost:9092',
        },
        topic: 'delay_service',
        groupId: 'delay_service_1',
        topicWithOrder: 'delay_service_order',
        groupIdWithOrder: 'delay_service_order_1',
        topicRetry: 'delay_service_retry',
        groupIdRetry: 'delay_service_retry_1',
    };
```

`database/config.json`

此文件的配置应该与`config/config.env.js`中的`sequelize`配置保持一致

`config.duties = [];`

此配置为项目的职责，传多个值则为多种职责，可做消息队列的消费和生产的分离，也可以将多个消费者分别部署

总共五种职责:

```
   DUTY_CONSUMER: 'consumer', // 普通无顺序请求的消费者
   DUTY_PRODUCER: 'producer', // 生产者，负责接收http请求
   DUTY_CONSUMER_RETRY: 'consumerRetry', // 重试消费者
   DUTY_CONSUMER_WITH_ORDER: 'consumerWithOrder', 有顺序请求的消费者
   DUTY_RETRY_PRODUCER: 'retryProducer', // 重试的生产者，定时任务负责将errorLogs的数据生产到消息队列去
```