# pragmatic-mq

**Repository Path**: wizard-lee/pragmatic-mq

## Basic Information

- **Project Name**: pragmatic-mq
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-10-14
- **Last Updated**: 2026-10-20

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# pragmatic-mq

**可以内嵌进应用的分布式消息队列 —— 一个依赖接入，只需 MySQL。**

消息队列通常是"先部署一套集群，再把应用接上去"。pragmatic-mq 把顺序反了过来：

| 差异点 | 说明 |
|---|---|
| **一个依赖就够** | 引入 `pragmatic-mq-starter` + 几行配置，MQ 随应用启动，没有独立进程要部署、监控、扩容 |
| **只有一个外部依赖** | MySQL。**没有 ZooKeeper，也没有独立的注册中心集群** |
| **同一套代码两种形态** | 开发测试用嵌入式、生产用独立部署，业务代码**一行不改** |
| **不污染你的依赖** | gRPC 的内嵌 Netty 已整体重定位，**不产生任何 `io.netty` 传递依赖**，与业务方自己的 Netty 不冲突 |
| **注解式消费** | 加个 `@MqListener` 就能消费，自动发现、注册、随应用启动 |

需要跨进程时把 `mode` 从 `embedded` 改成 `cluster`、指向 Name Server 即可 —— 生产者、消费线程模型、offset 管理、重试、注解式消费**全部复用同一份实现**，因为两种模式的分界只有一处：`MqApiProvider` 的实现选择。

```mermaid
flowchart LR
    subgraph EMBEDDED["嵌入式 mode=embedded"]
        direction TB
        EB["业务代码<br/>@MqListener / producer.send()"] --> EC["MqClient"]
        EC --> EL["LocalApiProvider"]
        EL --> EBK["进程内 Broker Bean"]
    end

    subgraph CLUSTER["独立部署 mode=cluster"]
        direction TB
        CB["业务代码<br/>同一份代码"] --> CC["MqClient"]
        CC --> CT["TransportApiProvider"]
        CT --> DNS["Name Server"]
        CT --> CBR["远端 Broker"]
    end

    EBK --> DB[("MQ 数据库")]
    DNS --> DB
    CBR --> DB
```

---

## 能力

| 能力 | 说明 |
|---|---|
| 传输可插拔 | gRPC（默认）/ HTTP，通过 SPI 发现，加传输只需加模块 |
| 队列 + Offset 模型 | 一个 topic 多队列，offset 按队列推进，支持多消费者并发消费 |
| 服务端分配队列 | 由 Broker 在存活线程间轮转分配，多实例消费同一 topic 不会打架 |
| 动态扩缩容 | 依据 **Broker 视角**的存活线程数与队列数比较后调整，避免多实例各自扩张 |
| 失败重试 | 消费失败转入 Broker 侧重试通道（按源队列分区、按失败应用隔离），客户端按队列拉取到期重试并逐条确认；延迟 `n * 5s` 递增，超限进死信 |
| 定时消息 | 到点后由**消费该队列的客户端驱动** Broker 转入普通通道（按队列属主触发，无消费者则不搬迁） |
| 线程心跳与回收 | 客户端续租，Broker 回收超时线程并重新分配队列 |
| 幂等 | 消息 `uuid` 由 Broker 生成，重复投递不产生重复消息 |

---

## 适用性

### 适合

- **同一套代码要跑两种形态**：开发与测试用嵌入式（不依赖任何外部进程），生产用独立部署，业务代码一行不改；
- **已有 MySQL，不想为消息再引入一个中间件集群**——不需要 ZooKeeper，也不需要独立的 Name Server 集群；
- **应用之间需要可靠投递**：订单异步化、事件通知、失败重试、延时任务；
- **希望"少一个运维组件"**：嵌入式模式随应用启动，没有独立进程要部署、监控、扩容。

### 不适合

| 场景 | 原因 |
|---|---|
| 百万级 TPS | 存储是单库 MySQL，拉取是轮询语义，未做分片、零拷贝或批量聚合 |
| 事务消息 / 半消息 | 未实现 |
| 多副本强一致与自动故障转移 | 选主是基于 `br_get_master` 表的**租约**（过期即可抢占），不是 Raft 之类的共识算法 |
| Kafka 式的分区有序 + 消费组 rebalance 语义 | 模型更简单：**队列 + offset**，有序性只在单队列内成立 |
| 多租户与细粒度权限 | 无 |
| 非 MySQL 数据库 | **仅支持 MySQL**，可放心使用 MySQL 特有语法 |

### 设计取舍

| 方面 | 选择 | 含义 |
|---|---|---|
| 投递模型 | **Pull** | 消费者主动拉取，节奏由客户端掌握 |
| 拉取方式 | **同步循环** | 线程数 ≥ 队列数时一个线程一个队列，无串行损失；不引入独立 IO 线程池 |
| 队列分配 | **服务端决定** | 多实例消费同一 topic 不会互相抢队列 |
| Offset | 客户端提交、Broker 存储，**按队列粒度** | 批量提交，重启不重放 |
| 失败重试 | **Broker 侧** | 可持久化、可隔离到失败应用、超限进死信 |
| 存储 | **MySQL** | 复用既有运维能力，代价是吞吐上限 |
| 选主 | **表租约** | 简单，适合中小规模 |
| 传输依赖 | **shaded Netty** | 不与业务方的 Netty 冲突 |

---

## 模块

| 模块 | 职责 |
|---|---|
| `pragmatic-mq-protocol` | 契约层：`ProducerApi` / `ConsumerApi` / `NameServerApi`、模型、Protobuf 定义、传输 SPI。**纯 Java，无 Spring** |
| `pragmatic-mq-transport-grpc` | gRPC 传输（客户端 + 服务端绑定层） |
| `pragmatic-mq-transport-http` | HTTP + JSON 传输 |
| `pragmatic-mq-server` | 服务端：存储访问 + Broker 消息通道 + Name Server 元数据路由（原 storage / broker / nameserver 三模块合并；Java 包保持 `io.pragmatic.mq.storage` / `broker` / `nameserver` 分立，按配置类分角色装配） |
| `pragmatic-mq-client` | 客户端 SDK：生产者、消费者、服务发现、`@MqListener` |
| `pragmatic-mq-admin-client` | 管理端契约：`NameServerAdminApi` / `BrokerAdminApi`（控制台 / CLI / 运维脚本用） |
| `pragmatic-mq-admin-transport` | 管理面传输 SPI：客户端工厂 / 服务端绑定 / 传输配置（纯 Java，无 Spring） |
| `pragmatic-mq-admin-transport-grpc` | 管理面 gRPC 传输（默认）：proto 桩、客户端代理、服务端绑定、Broker 寻址 |
| `pragmatic-mq-admin-transport-http` | 管理面 HTTP/JSON 传输（可选）：JDK HttpServer + Jackson |
| `pragmatic-mq-admin-web` | 可集成 REST 包：Topic / 关系 / 消息查看 / 监控 / 消息管理 + 令牌鉴权 |
| `pragmatic-mq-admin` | 独立部署管理后台应用壳（`apps` profile，零数据源） |
| `pragmatic-mq-starter` | Spring Boot 自动装配：一个依赖接入嵌入式模式 |
| `pragmatic-mq-bootstrap` | 独立部署启动入口，按角色启动 |

POM 分三层：根 `pragmatic-mq` 只做聚合构建、**不发布**；`pragmatic-mq-parent` 管第三方版本与构建配置，`pragmatic-mq-bom` 管库模块的版本，两者都发布。`bootstrap` / `admin` / `examples` 只在 `-Papps` 下构建，不进入仓库。

依赖方向（Maven 会自动排序，此表用于核对）：

```text
protocol ─┬─> transport-grpc ─┐
          ├─> transport-http ─┤
          ├─> server ─────────┼─> starter ─> bootstrap
          ├─> client ─────────┘
          └─> admin-client ─┬─> admin-transport ─┬─> admin-transport-grpc ─┐
                            │                     └─> admin-transport-http ─┤
                            └─> admin-web ──────────────────────────────────┴─> admin
```

分层约束：`protocol` / `transport-*` / `admin-client` / `admin-transport*` **不依赖 Spring**（纯 Java + Protobuf / Jackson）；Spring 只出现在 `server` / `starter` / `bootstrap` / `admin-web` / `admin`。

---

## 快速开始

### 环境要求

| 项 | 版本 |
|---|---|
| JDK | 17+ |
| MySQL | 5.7+ / 8.x（仅支持 MySQL，可放心使用 MySQL 特有语法） |
| Maven | 3.8+ |

### 建表

```bash
mysql -uroot -p < pragmatic-mq-server/sql/nameserver_schema.sql
mysql -uroot -p < pragmatic-mq-server/sql/broker_schema.sql
```

依赖 `pragmatic-mq-starter` 后，两份脚本也在 jar 内：`classpath:META-INF/pragmatic-mq/sql/`。

### 构建

```bash
mvn clean install
```

构建会执行集成测试，因此**需要一个可用的 MySQL**：默认连 `127.0.0.1:13306` 的 `pragmatic_mq_test` 库。没有数据库时请跳过测试：

```bash
mvn clean install -DskipTests
```

默认只构建 parent、BOM 与 9 个库模块。需要 `bootstrap` / `examples` 时加 `-Papps`：

```bash
mvn clean install -DskipTests -Papps
```

指向别的库用环境变量 `MQ_TEST_DB_URL` / `MQ_TEST_DB_USERNAME` / `MQ_TEST_DB_PASSWORD`。

### 方式一：嵌入式（业务应用内直连）

引入 **BOM + 一个依赖**：

```xml
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.pragmatic-ddd</groupId>
            <artifactId>pragmatic-mq-bom</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>com.pragmatic-ddd</groupId>
        <artifactId>pragmatic-mq-starter</artifactId>
    </dependency>
</dependencies>
```

配置：

```yaml
pragmatic:
  mq:
    mode: embedded          # 默认值，可省略
    client:
      app: order-service    # 必填：消费隔离键 + 发布关系键
    embedded:               # 声明式 topic 引导：启动时幂等确保存在并注册发布关系
      topics:
        - name: order.created
          queue-count: 4
    datasource:
      url: jdbc:mysql://127.0.0.1:3306/pragmatic_mq?useSSL=false
      username: mq
      password: ******
```

topic 管理两种方式：**yml 声明**（如上，启动时幂等 ensure，多实例部署安全——各实例声明一致即可）或**程序化**——注入 `NameServerAdminApi`（createTopic / registerPublication / registerSubscription / updateComment / updateQueueCount / deleteTopic / queryBrokers），嵌入式下是进程内 Bean。`NameServerApi` 是运行时契约，只做发现（路由查询 / Topic 目录 / Broker 注册），**不写元数据**。topic 创建的完整说明见 `docs/Topic创建与消息收发指南.md`。

消费——加个注解就够了，starter 会扫描、注册并随应用启动：

```java
@Component
public class OrderConsumer {

    @MqListener(topic = "order.created")
    public void onOrder(Message message) {
        // 处理成功即推进 offset；抛异常则消息转入 Broker 的重试通道
    }
}
```

发送：

```java
@Service
public class OrderService {

    private final MessageProducer producer;

    public OrderService(MqClient client) {
        this.producer = client.producer();
    }

    public void placeOrder(String payload) {
        producer.send(List.of(OutgoingMessage.of("order.created", payload, "biz-1")));
    }
}
```

> **嵌入式不监听任何端口**，外部进程无法连接。需要外部消费者时，请起独立的 broker 进程（见方式二）。

### 方式二：独立部署

**启动 Name Server**

```bash
java -jar pragmatic-mq-bootstrap/target/pragmatic-mq-bootstrap-1.0.0-SNAPSHOT.jar \
     --spring.profiles.active=nameserver
```

**启动 Broker**

```bash
java -jar pragmatic-mq-bootstrap/target/pragmatic-mq-bootstrap-1.0.0-SNAPSHOT.jar \
     --spring.profiles.active=broker
```

`-Drole=broker` 是等价写法（内部映射为 profile）；`broker,nameserver` 可在单进程内同时启动两个角色。

**客户端接入**——独立部署不提供 `client-starter`，业务方自行组装 `client + transport-grpc`：

```xml
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.pragmatic-ddd</groupId>
            <artifactId>pragmatic-mq-bom</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>com.pragmatic-ddd</groupId>
        <artifactId>pragmatic-mq-client</artifactId>
    </dependency>
    <dependency>
        <groupId>com.pragmatic-ddd</groupId>
        <artifactId>pragmatic-mq-transport-grpc</artifactId>
    </dependency>
</dependencies>
```

```java
MqClient client = DefaultMqClient.remote(
        MqClientProperties.remote("order-service", "grpc://nameserver-host:9501"));

client.registerAnnotated(listenerBean);   // 扫描该 bean 上的 @MqListener
client.start();

// 应用关闭时
client.shutdown();
```

`MqClientProperties.remote(...)` 默认使用 gRPC；换 HTTP 时额外依赖 `pragmatic-mq-transport-http` 并指定 transport。业务方自己负责 `MqClient` Bean 的创建、监听器的收集与启动 / 关闭时机——一个 `@Configuration` + `SmartLifecycle` 即可。

### 日志

日志用 **Logback**（`logback-classic`，版本随 Spring Boot BOM），配置在 bootstrap 的 `src/main/resources/logback-spring.xml`：

- **双输出**：控制台（彩色单行）+ 文件 `logs/pragmatic-mq.log`（UTF-8）
- **滚动归档**：按天 + 单片 10MB（`SizeAndTimeBasedRollingPolicy`），保留 30 天、总量上限 2GB
- **默认级别**：root `INFO`、`io.pragmatic.mq` `DEBUG`（队列分配、位点、重试、定时搬迁、路由排障直接可查）；`io.grpc` / `org.mybatis` 降为 `WARN` 降噪
- **免改文件覆盖**：`LOG_PATH=/var/log/mq` 换日志目录；`--logging.level.io.pragmatic.mq=INFO` 临时调级别；Docker 部署默认 `LOG_PATH=/app/logs`，compose 已挂载 `pragmatic-mq-logs` 卷

代码层统一 SLF4J API（`LoggerFactory.getLogger`），实现由宿主应用提供——纯 Java 模块（protocol / transport / client 等）不带后端，其单测通过 test 作用域的 `logback-classic` 保留日志输出。

---

## 管理后台

`pragmatic-mq-admin-web` 提供一套 REST 接口（只到接口层，不含前端与审计），支持两种形态。

### 形态一：嵌入式（随业务应用）

引入依赖即可，`AdminWebAutoConfiguration` 检测到进程内的 `NameServerAdminApi` + `BrokerAdminApi` 后自动注册控制器：

```xml
<dependency>
    <groupId>com.pragmatic-ddd</groupId>
    <artifactId>pragmatic-mq-admin-web</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>
```

```yaml
pragmatic:
  mq:
    admin:
      context-path: /mq-admin   # 默认值
      token: change-me          # 请求头 X-Mq-Admin-Token；留空关闭校验
```

### 形态二：独立部署（零数据源）

1. 集群节点开启 admin 传输服务端（默认关闭）：

```yaml
pragmatic:
  mq:
    admin:
      transport:
        enabled: true
        type: grpc
        broker-port: 9001        # 运行时端口仍为 9000
        name-server-port: 9502   # 运行时端口仍为 9501
```

2. 构建并启动管理后台（`-Papps` 下产出 `pragmatic-mq-admin`）：

```bash
mvn -q -Papps clean install -DskipTests
java -jar pragmatic-mq-admin/target/pragmatic-mq-admin-1.0.0-SNAPSHOT.jar
```

```yaml
pragmatic:
  mq:
    admin:
      token: change-me
      transport:
        type: grpc
        addressing: route        # route（默认）| lb | static
        name-server-host: nameserver-host
        name-server-port: 9502
        broker-port: 9001
server:
  port: 9600
```

`route` 经 `NameServerAdminApi.queryBrokerNodes()` 取存活节点清单，按 node id 轮询，连接失败在剩余节点内重选一次；`lb` / `static` 为退化形态（配置 `lb-broker` / `static-broker`）。传输可选 `http`（引入 `pragmatic-mq-admin-transport-http`）。

### REST 端点

| 分组 | 路径 |
|---|---|
| Topic | `POST/GET /api/topics`、`PUT /api/topics/{name}/queue-count`、`POST .../reset-queue-count`、`PUT .../comment`、`DELETE /api/topics/{name}` |
| 关系 | `POST/DELETE/GET /api/publications`、`/api/subscriptions` |
| 消息查看 | `GET /api/timing-messages`、`/api/timing-messages/{id}`、`/api/retry-messages`、`/api/retry-messages/{id}` |
| 监控 | `GET /api/monitor/overview`、`/backlog`、`/dead-letters`、`/brokers` |
| 消息管理 | `DELETE /api/retry-messages/{id}`、`POST /api/retry-messages/{id}/redeliver` |

错误统一为 `{"code":"...","message":"..."}`；分页 `page`（从 0 起）/ `size`（上限 100）。积压与死信重投为两段式编排：先经 NameServer 取 Topic 定义，再调 Broker；topic 缺失返回 404。

---

## 配置

所有配置键统一使用 `pragmatic.mq` 命名空间。

### 客户端 / 通用

| 配置 | 默认 | 说明 |
|---|---|---|
| `pragmatic.mq.mode` | `embedded` | `embedded` / `cluster` |
| `pragmatic.mq.client.app` | **必填** | 应用名 |
| `pragmatic.mq.client.transport` | `grpc` | `grpc` / `http` |
| `pragmatic.mq.client.name-server-url` | — | `cluster` 模式必填，缺失则启动失败 |
| `pragmatic.mq.client.min-threads` | `1` | 单 topic 消费线程下限 |
| `pragmatic.mq.client.max-threads` | `30` | 单 topic 消费线程上限 |
| `pragmatic.mq.client.heartbeat-interval` | `10s` | 线程心跳周期，须远小于 broker 的 `thread-timeout` |
| `pragmatic.mq.client.assignment-ttl` | `5s` | 队列分配缓存有效期 |
| `pragmatic.mq.client.request-timeout` | `10s` | 单次请求超时 |

需要 `MqProperties` 未覆盖的旋钮时，直接定义自己的 `MqClientProperties` Bean——starter 的 Bean 会退让。

### 服务端

| 配置 | 默认 | 说明 |
|---|---|---|
| `pragmatic.mq.server.transport` | `grpc` | 节点暴露的协议，`grpc` / `http` 二选一 |
| `pragmatic.mq.server.port` | `9000` / `9501` | broker / nameserver 监听端口 |
| `pragmatic.mq.broker.*` | 见 `BrokerProperties` | broker-id、线程超时、重试延迟等 |
| `pragmatic.mq.nameserver.*` | 见 `NameServerProperties` | broker 超时、回收周期等 |
| `pragmatic.mq.datasource.*` | — | url / username / password / 连接池参数 |

> `pragmatic.mq.datasource` 是 MQ 专属数据源，**不与业务数据源共用**，Bean 也不标 `@Primary`，可与业务数据源在同一应用内共存。

---

## 传递依赖与冲突处理

`pragmatic-mq-bom` 只固定 `pragmatic-mq-*` 模块的版本，**不接管业务方的第三方依赖**。SDK 会用到的第三方如下，冲突时请在业务方自己的 `dependencyManagement` 里收敛：

| 依赖 | 来源 | 处理方式 |
|---|---|---|
| Spring Boot 3.2.4 | `pragmatic-mq-parent`（BOM 继承，import 后可见） | 业务方自己的 Boot BOM 优先；把它的 import 写在 `pragmatic-mq-bom` 之前即可覆盖 |
| gRPC 1.60.0 / protobuf 3.25.1 | `transport-grpc`、`protocol` | 两者版本必须匹配，升级时一起升 |
| Guava 32.1.3-jre | `protobuf-java`、`grpc-*` | 业务方已有 Guava 时显式锁定版本 |
| MyBatis 3.5.16 / mybatis-spring 3.0.3 | `storage`（仅嵌入式） | 不再传递 Boot starter，不会触发 `MybatisAutoConfiguration` |
| HikariCP、MySQL driver | `spring-boot-starter-jdbc`、`storage` | 嵌入式必需；独立部署不引入 |
| Jackson | `transport-http`、`storage` | 由 Spring Boot BOM 管；只用 gRPC 的独立部署不会经由 `transport-http` 引入 |

Netty 不在此列：`grpc-netty-shaded` 已把 Netty 重定位，不产生 `io.netty` 传递依赖。

---

## 项目状态

**已完成**：阶段 0~7 —— 协议层、存储层、Broker、Name Server、客户端 SDK、Starter 与启动引导。

**未完成**：阶段 8（测试、验证与交付）—— 接入示例工程、运维说明、性能基线、版本发布。当前版本为 `1.0.0-SNAPSHOT`，**尚未作为正式版本交付**。

---

## 许可

[Apache License 2.0](LICENSE)

```text
Copyright 2026 wizard-lee

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
```