eleven26

在工作中，往往有一些需要对大批量数据进行分析的场景，依赖于传统关系型数据库（如 MySQL）的话，效率往往及其低下。 相比之下，一些 OLAP 数据库在统计分析等场景下表现优异，比如 ClickHouse。

比如：

1. 扫描数据量：ClickHouse 采用列式存储，同一列的数据物理上连续存储，查询时只需读取目标列，而 MySQL 查询一列需要扫描 20 列数据（假设表有 20 列）
2. 数据压缩：ClickHouse 存储的数据是经过压缩的，读取的时候可以减少 IO 的开销。
3. 多核计算：ClickHouse 的查询任务可以自动拆分到多个 CPU 核心执行，充分利用现代硬件并行能力。

也就是说，不管是存储、读取还是计算，专门的 OLAP 数据库都比 OLTP 关系型数据库要高效。 正因如此，我们会选择使用一些 OLAP 数据库来处理一些数据分析的业务，比如 ClickHouse，这个时候如何将数据从 MySQL 同步到 ClickHouse 就成了一个亟需解决的问题。

如何将数据从 MySQL 同步到 ClickHouse

目前比较有效的在 MySQL 和 ClickHouse 之间同步数据的方式都是通过 binlog 捕获数据库变更，然后写入到 ClickHouse。比如有如下的一些方案：

1. Debezium + Kafka：通过 Debezium 捕获 MySQL 的 Binlog 变更事件，经 Kafka 中转后写入 ClickHouse。
2. Canal + Kafka：通过 Canal 捕获 MySQL 的 Binlog 变更事件，经 Kafka 中转后写入 ClickHouse。
3. 使用阿里云 DTS（数据传输服务）：DTS 支持 MySQL 到 ClickHouse 的实时同步。

但是直接通过捕获 MySQL 的 binlog 写入 ClickHouse，在运维层面复杂度还是偏高，如果使用的是阿里云的 RDS， 可以直接使用阿里云的 DTS 来进行 MySQL 到 ClickHouse 的同步，好处是我们不需要处理底层同步的一些细节问题。

DTS 里面包含了几种服务，本文只是使用它提供的数据订阅服务。

阿里云 DTS 数据订阅的工作原理

数据订阅支持实时拉取 RDS 实例的增量日志，用户可以通过 DTS 提供的 SDK 数据订阅服务端来订阅增量日志，同时可以根据业务需求，实现数据定制化消费。

在这个过程中，我们需要处理的就是 业务逻辑 部分，也就是获取到 RDS 的数据库变更之后，如何将这些变更写入到 ClickHouse。我们可以选择处理全部字段，或者只处理自己需要的某些字段。

如何使用阿里云 DTS 同步 MySQL 到 ClickHouse

具体文档可在阿里云文档查看：数据传输服务 - 快速入门 - 数据订阅操作指导。

ClickHouse 安装配置（CentOS）

我们可以通过如下命令来安装 ClickHouse：

1
2
3

sudo yum install -y yum-utils
sudo yum-config-manager --add-repo https://packages.clickhouse.com/rpm/clickhouse.repo
sudo yum install -y clickhouse-server clickhouse-client

安装完毕后，通过下面的命令来启动 ClickHouse 服务：

1
2
3

sudo systemctl enable clickhouse-server
sudo systemctl start clickhouse-server
sudo systemctl status clickhouse-server

如果一切顺利，我们会看到如下输出：

1
2
3
4
5
6
7

● clickhouse-server.service - ClickHouse Server (analytic DBMS for big data)
   Loaded: loaded (/usr/lib/systemd/system/clickhouse-server.service; enabled; vendor preset: disabled)
   Active: active (running) since Mon 2025-03-17 09:19:03 CST; 2 weeks 1 days ago
 Main PID: 2865 (clickhouse-serv)
    Tasks: 822
   Memory: 1.3G
   CGroup: /system.slice/clickhouse-server.service

如果不是绿色的 active，则可以通过 journalctl -xe -u clickhouse-server 查看一下具体错误。

下面是通过 yum install 安装的 ClickHouse 的一些目录：

• /var/log/clickhouse-server 日志
• /var/run/clickhouse-server 运行时的一些文件
• /etc/clickhouse-server 配置目录
• /var/lib/clickhouse

配置文件位置：/etc/clickhouse-server/config.xml，我们可以通过修改这个配置文件来调整 ClickHouse 服务的相关参数。如：

• listen_host 修改绑定的 ip 地址，如果需要给本机以外的人访问，可以修改为 0.0.0.0

连接到 ClickHouse

我们可以通过下面的命令来连接到 ClickHouse：

1

clickhouse-client --host 127.0.0.1 --port 9011

这跟 MySQL 的命令行类似，除了样子长得像以外，很多的 SQL 语句都是跟 MySQL 一样的。比如，进入交互式的命令行之后，我们可以通过 show databases; 来查看有哪些数据库：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

➜  ~ clickhouse-client --host 127.0.0.1 --port 9011
ClickHouse client version 25.2.1.3085 (official build).
Connecting to 127.0.0.1:9011 as user default.
Connected to ClickHouse server version 25.2.1.

Warnings:
 * Linux transparent hugepages are set to "always". Check /sys/kernel/mm/transparent_hugepage/enabled

example.com :) show databases;

SHOW DATABASES

Query id: 27bcb695-878d-4f0d-8a6a-f3b07fa0d498

   ┌─name───────────────┐
1. │ INFORMATION_SCHEMA │
2. │ default            │
3. │ information_schema │
4. │ system             │
   └────────────────────┘

4 rows in set. Elapsed: 0.005 sec.

注意：这里的 9011 端口是我本地的端口，不是默认的端口。

创建对应的 ClickHouse 数据库和表

现在，假设我们的 MySQL 中有一个名字为 order 的数据库，里面有一个 orders 的表。然后我们需要将其中的一些字段同步到 ClickHouse 中，首先我们需要创建对应的数据库和表。

1. 创建数据库

1

create database order;

2. 创建表

1
2
3
4
5
6
7
8
9
10
11
12
13

create table `orders`(
	id UInt64,
	company_id UInt64,
	supplier_id UInt64,
	order_total_fee Decimal(20, 4),
	coupon_id Nullable(UInt64),
	coupon_price Nullable(Decimal(20, 4)),
	deleted_at Nullable(Datetime),
	created_at Datetime,
	PRIMARY KEY (created_at, id)
) ENGINE = MergeTree()
PARTITION BY toStartOfMonth(created_at)
ORDER BY (created_at, id);

初次同步数据到 ClickHouse

因为 DTS 只能捕获增量数据，初次同步的时候我们需要手动将 MySQL 中的全量数据导入到 ClickHouse。如果表的数据写入频繁的话，可能需要停服之后再进行此操作。

我们可以通过下面的命令来从 MySQL 直接导数据到 ClickHouse：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

INSERT INTO orders SELECT
id,
company_id,
supplier_id,
order_total_fee,
coupon_id,
coupon_price,
deleted_at,
created_at 
FROM
	mysql (
		'127.0.0.1:3306',
		'order',
		'orders',
	    '账号',
	    '密码')

在实际执行中，将 MySQL 里面的信息替换为你自己的 MySQL 配置即可。在导入数据完成后，可以通过下面的命令来查看导入是否成功：

1
2
3
4
5
6
7
8
9
10
11
12
13

example.com :) select count(*) from orders;

SELECT count(*)
FROM orders

Query id: 227ce057-5949-4d33-b5ae-2f2ce7e96e38

   ┌─count()─┐
1. │   25781 │
   └─────────┘

1 row in set. Elapsed: 0.039 sec. Processed 25.78 thousand rows, 103.13 KB (668.04 thousand rows/s., 2.67 MB/s.)
Peak memory usage: 401.44 KiB.

如果 count 的值跟 MySQL 中的数据一致，则说明导入成功。

在这一步执行完成之后，就可以启动 DTS 的数据订阅服务了。后续的数据库变动会被捕获到，直到我们 ack 为止。

MySQL 跟 ClickHouse 的字段对应关系

太长了，自己复制到浏览器打开：

https://help.aliyun.com/zh/dts/user-guide/use-a-kafka-client-to-consume-tracked-data-2#section-woc-4pq-mes

编写数据订阅的代码（Python）

官方 demo 在：https://github.com/aliyun/aliyun-dts-subscribe-demo/tree/master/python

核心代码如下（比较标准的 Kafka 消费者代码）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

dts_record_schema = schema.load_schema(
    "dts_record_avsc/com.alibaba.dts.formats.avro.Record.avsc"
)

def decode(msg_value):
    message_bytes = io.BytesIO(msg_value)
    data = schemaless_reader(message_bytes, dts_record_schema)
    return data

if __name__ == "__main__":
    try:
        # Kafka Consumer 配置参数
        topic_name = "cn_hangzhou_vpc_rm_bp1907x8zbo20z60u_dts_upgrade_from_old_version2"
        auto_commit = False
        # 消费组 ID
        group_id = "dtse9gh4883283o991"
        sasl_mechanism = "PLAIN"
        security_protocol = "SASL_PLAINTEXT"
        username = "xiaqiutest"
        password = "DTStest1234"
        bootstrap_servers = ["dts-cn-hangzhou.aliyuncs.com:18001"]

        # 如果username不含有group_id，则更新username为username-group_id
        if group_id not in username:
            username = username + "-" + group_id

        # 创建 KafkaConsumer 实例
        consumerGroupHandler = KafkaConsumer(
            topic_name,
            enable_auto_commit=auto_commit,
            group_id=group_id,
            sasl_mechanism=sasl_mechanism,
            security_protocol=security_protocol,
            sasl_plain_username=username,
            sasl_plain_password=password,
            bootstrap_servers=bootstrap_servers,
        )

        print("start")
        for msg in consumerGroupHandler:
            record = decode(msg.value)
            # import datetime

            # sourceTimestamp = record.get("sourceTimestamp")
            # formatted_date = datetime.datetime.fromtimestamp(sourceTimestamp).strftime(
            #     "%Y-%m-%d %H:%M:%S"
            # )
            # print(formatted_date)
            print(record)

        print("end")
    except Exception as e:
        print(e)
        traceback.print_exc()

但是官方的 demo 实在是过于简陋，所以这里会针对里面一些比较关键的地方说明一下：

• schema.load_schema 和 schemaless_reader，python 版本的 SDK 使用了 fastavro 来解析 Avro 格式的数据，所以需要用到 fastavro.schemaless_reader 来读取数据。
• Kafka Consumer 配置参数在 DTS 订阅任务那里都可以看到，其中 username 和 password 是新建订阅任务的时候自己填写的。
• record = decode(msg.value) 这一行会将 Kafka 中的数据解码成字典格式，这样就可以直接使用了。
• record 里面包含的字段可以看 dts_record_avsc/com.alibaba.dts.formats.avro.Record.avsc 这里面的定义。
• operation 字段表示的是操作类型，比如 INSERT, UPDATE, DELETE 等。我们可以根据这个字段来决定在 ClickHouse 中执行什么样的操作。
• record 的 beforeImages 是数据变更前的字段值，afterImages 是数据变更后的字段值。在执行 insert 操作的时候 beforeImages 是空的，在执行 delete 操作的时候 afterImages 是空的。update 的时候，两个都有值，分别是变动前后的字段变动。

官方 demo 的代码只是给出了基本的消费的代码，在实际开发中，我们还需要根据 operation 字段来决定在 ClickHouse 中执行什么样的操作。如，我们可以像下面这样做一些插入、删除的操作：

1
2
3
4
5
6
7

if op == 'INSERT':
    self.insert(...)
if op == 'DELETE':
    self.clickhouse_client.delete(...)
if op == 'UPDATE':
    self.clickhouse_client.delete(...)
    self.insert(...)

更具体代码自行实现了。需要注意的是，在插入之前，可能需要做一些数据类型转换的操作，否则可能在插入到 ClickHouse 的时候出现类型不匹配的问题。

部署（CentOS）

本文以 conda 和 supervisor 来作为示例部署。

Python 环境（conda）

由于 Python 各版本的差异，所以更推荐使用 conda 来管理 Python 环境，我们可以通过以下命令来安装一下 miniconda：

1
2

wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
bash ~/Miniconda3-latest-Linux-x86_64.sh

安装到默认的 ~/miniconda3 目录即可。

conda 的常用命令：

1
2

/root/miniconda3/bin/conda env list # 列出所有环境
/root/miniconda3/bin/conda create --name dts python=3.11 # 创建 py 环境

更详细的内容可参考官方文档：https://www.anaconda.com/docs/getting-started/miniconda/install#macos-linux-installation

supervisor

1
2
3
4
5
6
7
8
9
10

[program:dts_consumer]
command=/root/miniconda3/envs/dts/bin/python main.py
numprocs=1
autostart=true
autorestart=true
startretries=3
user=root
redirect_stderr=true
directory=/home/www/dts-consumer
stdout_logfile=/var/log/dts_consumer.log

保存到服务器的 /etc/supervisord.d 目录，然后执行：

1

supervisorctl update

即可启动。

参考文档

• DTS 常见问题 https://help.aliyun.com/zh/dts/support/faq#section-fle-1lu-lla
• Kafka https://kafka.apache.org/intro
• fastavro https://fastavro.readthedocs.io/en/latest/
• 排查订阅任务问题 https://help.aliyun.com/zh/dts/user-guide/troubleshoot-issues-in-change-tracking-tasks
• MySQL 字段类型与 dataTypeNumber 数值的对应关系 https://help.aliyun.com/zh/dts/user-guide/use-a-kafka-client-to-consume-tracked-data-2#section-woc-4pq-mes

在数据迁移的时候，我们往往会有大量数据需要更新，如果使用 Eloquent 的 save 方法，那么每次更新都会执行一次 SQL 语句，这样会导致每一次更新都会有一次网络请求，数据量大的时候会非常慢。

MongoDB 本身提供了批量更新的方法，也就是使用 MongoDB\Driver\BulkWrite 这个类来实现批量更新，通过它我们就可以实现批量的更新操作了。

实现流程

以下是使用 MongoDB\Driver\BulkWrite 实现批量更新的流程：

1. 创建一个 BulkWrite 实例

1

$bulk = new MongoDB\Driver\BulkWrite(['ordered' => true]);

其中 ordered 参数表示是否按照插入的顺序执行操作，默认为 true。有序模式按操作顺序执行，失败则中断；无序模式允许并行执行，失败不影响后续操作。

若需严格保证插入顺序且不允许部分失败，使用 ordered: true；若需提高性能且允许部分失败，使用 ordered: false。

2. 添加更新操作

我们可以使用 BulkWrite 的 update 方法来添加更新操作，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12

foreach ($documents as $document) {
    $bulk->update(
        ['_id' => new ObjectId($document['_id'])],
        ['$set' => $this->prepareDocument($document)]
    );
}

private function prepareDocument(array $document): array
{
    unset($document['_id']); // 移除主键
    return $document;
}

其中，update 的第一个参数是查询条件，第二个参数是更新的数据。

在这个例子中，我们遍历了 $documents 数组，然后使用 update 方法来更新数据。

3. 执行批量更新

最后，我们可以使用 MongoDB\Driver\Manager 的 executeBulkWrite 方法来执行批量更新：

1
2
3

$manager      = new Manager('mongodb://localhost:27017');
$writeConcern = new WriteConcern(WriteConcern::MAJORITY, 30000);
$manager->executeBulkWrite('db0.user', $bulk, $writeConcern);

其中，executeBulkWrite 方法的第一个参数是数据库和集合名称，第二个参数是 BulkWrite 实例，第三个参数是 WriteConcern。

Manager 是 MongoDB 的连接管理器，WriteConcern 是写入关注级别，MAJORITY 表示大多数节点写入成功即可，30000 表示超时时间（也就是 30s）。

完整实现

如果我们使用了 Laravel 的 jenssegers/mongodb 扩展，那么我们可以直接使用如下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

<?php
use Jenssegers\Mongodb\Eloquent\Model;
use MongoDB\BSON\ObjectId;
use MongoDB\Driver\BulkWrite;
use MongoDB\Driver\Manager;
use MongoDB\Driver\WriteConcern;
use MongoDB\Driver\WriteResult;

class MongoBulkWriter
{
    /**
     * MongoDB 的模型
     *
     * @var Model
     */
    private $model;

    public function __construct(Model $model)
    {
        $this->model = $model;
    }

    /**
     * mongo 批量更新
     *
     * @param Model $model     MongoDB 的模型
     * @param array $documents 更新的 mongodb 数据，每个文档必须包含 _id 字段，以及需要更新的字段
     *
     * @return WriteResult
     */
    public static function write(Model $model, array $documents = []): WriteResult
    {
        $instance = new static($model);
        return $instance->doWrite($documents);
    }

    /**
     * mongo 批量更新
     *
     * @param array $documents 更新的mongodb数据
     *
     * @return WriteResult
     */
    private function doWrite(array $documents = []): WriteResult
    {
        assert(!empty($documents), '更新数据不能为空');
        assert(!empty($this->username()), 'MongoDB 用户名不能为空');

        $bulk = new BulkWrite(['ordered' => true]);

        foreach ($documents as $document) {
            assert(isset($document['_id']), '数据主键不能为空');

            $bulk->update(
                ['_id' => new ObjectId($document['_id'])],
                ['$set' => $this->prepareDocument($document)]
            );
        }

        // 包含特殊字符的密码的时候，需要通过第二个参数传递用户名和密码
        $manager      = new Manager($this->connectionUri(), ['username' => $this->username(), 'password' => $this->password()]);
        $writeConcern = new WriteConcern(WriteConcern::MAJORITY, 30000);
        return $manager->executeBulkWrite($this->namespace(), $bulk, $writeConcern);
    }

    private function namespace(): string
    {
        $database = $this->model->getConnection()->getConfig()['database'];

        return $database . '.' . $this->model->getTable();
    }

    private function connectionUri(): string
    {
        $connection = $this->model->getConnection();

        $config = $connection->getConfig();

        $mongoDbHost = $config['host'];
        $mongoDbPort = $config['port'];

        $database = $config['database'];

        return 'mongodb://' . $mongoDbHost . ':' . $mongoDbPort . '/' . $database;
    }

    private function username()
    {
        return $this->model->getConnection()->getConfig()['username'];
    }

    private function password()
    {
        return $this->model->getConnection()->getConfig()['password'];
    }

    private function prepareDocument(array $document): array
    {
        unset($document['_id']); // 移除主键
        return $document;
    }
}

关键说明：我们可以直接通过 MongoDB 的模型来获取其关联的数据库配置信息，然后用这些信息来建立起 MongoDB 连接，最后通过这个连接的 executeBulkWrite 方法来执行批量更新。

开始之前，我们来看看下面这一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public function couponCron() {
  $now = Carbon::now()->toDateTimeString();

  // 到开始时间了，未开始的优惠券活动更改为开始状态
  app(Coupon::class)
      ->where('coupon_status', 1)
      ->where('release_start_time', '<=', $now)
      ->update(['coupon_status' => 2]);
  
  // 过期优惠券活动
  $expireCouponIds = app(Coupon::class)
      ->whereIn('coupon_status', [2, 3])
      ->where('release_end_time', '<=', $now)
      ->pluck('id');
  
  if ($expireCouponIds) {
      // 过期优惠券活动状态改为失效
      app(Coupon::class)
          ->whereIn('id', $expireCouponIds)
          ->update(['coupon_status' => 4]);
  }
}

这段代码是一个定时任务，用于处理优惠券活动的状态变更：到了优惠券开始时间，将优惠券状态更改为活动中状态；到了优惠券结束时间，将优惠券状态更改为失效状态。

没有想到非常好的命名，就这样吧。

这段代码有什么问题？

像以上这个规模的代码，并没有什么问题，因为足够简单，我们看几秒注释就能知道这段代码的作用。 但是当这个函数几百行的时候，我们估计会比较头疼了，比如下面这种六七百行的：

单单比较长度可能看不出什么猫腻，我们再看看下面的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

public function couponCron()
{
    $this->startCoupons();
    $this->invalidExpiredCoupons();
}

private function startCoupons(): void
{
    $now = Carbon::now()->toDateTimeString();

    app(Coupon::class)
        ->where('coupon_status', 1)
        ->where('release_start_time', '<=', $now)
        ->update(['coupon_status' => 2]);
}

private function invalidExpiredCoupons(): void
{
    $now = Carbon::now()->toDateTimeString();

    $expireCouponIds = app(Coupon::class)
        ->whereIn('coupon_status', [2, 3])
        ->where('release_end_time', '<=', $now)
        ->pluck('id');

    if ($expireCouponIds) {
        app(Coupon::class)
            ->whereIn('id', $expireCouponIds)
            ->update(['coupon_status' => 4]);
    }
}

跟前一个例子不同之处在于：

1. 将代码拆分成了两个函数，每个函数只做一件事。
2. 原来的那个函数里面只需要调用两个函数即可。

只是对数据库操作做了一下封装，是这样吗？

从表面来看，是的，只是对数据库操作做了一下封装。但是从更深层次来看，也没有那么简单，我们将业务逻辑和技术细节分离开了：startCoupons 以及 invalidExpiredCoupons 是直接告诉读者业务是什么，而 startCoupons 函数的实现则告诉读者如何实现这个业务。 在我们只需要了解大概业务逻辑的情况下，无疑第二种写法能够更快地让我们理解代码，只有当需要去对其中某些业务逻辑做变更的时候，我们才需要去看具体的实现。 函数过长的一个原因往往就是，代码全都是描述如何实现业务逻辑的技术细节，而不是描述业务逻辑本身。

另外一种说法是：抽象层次不一样的代码混在一起，可能有一部分代码是描述业务逻辑的，然后一部分代码是描述实现业务逻辑的细节的，描述实现细节的代码我们需要看注释才能知道背后的业务逻辑。

在《重构》中，对于长方法，有这么一句话：如果你觉得需要对方法中的某些内容进行注释，你应该把这段代码放在一个新的方法中。如果需要解释，即使是一行也可以而且应该拆分成一个独立的方法。如果方法具有描述性名称，则无需查看代码即可了解它的作用。

《重构》中没有说到的

其实对于上面的这种问题，可能很多人也有意识到，要将函数写短一些，但是这种意识可能并不能很好地指导我们的实践：每个人标准不一样，比如每个人对于方法的 “短” 的定义不一样，有的人觉得 100 行就很长了，有的人觉得 200 行也没问题，当然实际中也不可能只从方法长短来判断一个方法写得好不好。 那什么东西应该被用来指导我们去在适当的时候将方法写短一点呢？我觉得可以从以下几个角度重新审视一下开篇的那一段代码，然后我们就很容易理解了：

1. 抽象层次：这个方法是描述业务逻辑的，还是描述实现业务逻辑的细节的？如果我们的方法是一个 API 或者某个逻辑实现的主入口，那我们应该将这个方法写得尽可能短，然后将业务逻辑的实现细节放到其他方法中。
2. 代码可读性：方法太长之后，我们可能看几行代码就得想想这几行代码到底做了什么操作，对后面的代码有什么影响。这样导致的一个问题是，我们的思考过程不会连贯，而是断断续续的，这样会导致我们的思维负担加重。
3. 可维护性：需要花费更多的时间去理解这个方法，然后才能修改它。另外一个问题是，可能其中某些技术的细节是可以复用的，在其他地方如果也有类似的细节，我们可能会重复写一遍。

什么是业务逻辑？什么是技术细节？

举个简单的例子，有个用户注册的接口的逻辑是：

1. 通过账号密码注册，然后发送邮件给用户。
2. 为了防止黑客批量注册，注册的时候需要填写一个验证码。在注册之前，需要先验证验证码是否正确。

在这个例子，主要的业务逻辑有：

1. 验证：账号密码验证码是否为空、验证码是否正确、用户是否已经存在。
2. 注册：在数据库中创建一条用户记录。
3. 发送邮件：通过某个邮件组件发送邮件给用户，告知用户注册成功。

上面提到的这三点，冒号前面的就是业务逻辑，冒号后面的就是技术细节。对于整体的业务逻辑，我们的伪代码可以写成这样：

1
2
3
4
5
6
7
8
9

public function register($account, $password, $captcha)
{
    // 验证
    $this->validate($account, $password, $captcha);
    // 注册
    $this->createUser($account, $password);
    // 发送邮件
    $this->sendEmail($account);
}

而技术细节部分，我们只需要在各自的方法中实现即可。这样的好处是，自己或者别人再看这段代码的时候，看到的首先都是业务逻辑，而不是一上来就陷入到技术细节中无法自拔。

分离开业务逻辑与技术细节的好处

其实写代码跟写作有点类似，因为二者都是有点 “写作” 的味道的，当然写作的创作性更强；另外就是，写代码跟写作最终都会让被人阅读，所以我们写代码的时候，也要考虑到别人阅读的问题。

对受众（包括读者、听众、观众或学员）来说，最容易理解的顺序是：先了解主要的、抽象的思想，然后了解次要的、为主要思想提供支持的思想。因为主要思想总是从次要思想中概括总结得出，文章中所有思想的理想结构必定是一个金字塔结构-由一个总的思想统领多组思想 - 《金字塔原理》

当我们将业务逻辑与技术分离之后，“业务逻辑” 就等于是我们的 “主要思想”，而 “技术细节” 就等于是 “次要思想”。这样就能让读者很容易就能理解我们的代码。

我们该如何做好分离？

业务逻辑与技术细节耦合的最常见的一种场景是，在我们实现某个功能的时候，写到哪一步就做哪一步业务逻辑所需要的 CRUD 操作，当然可能不同语言、框架底下写出来的代码会有所不同，但是我们接触的大多数都是这样的代码。

当然并不是说这是错的，这是很正常的，实现功能的代码一定是长这个样子的，很少纯粹只有业务逻辑的代码，纯粹的技术的代码倒是有很多（比如 RPC 通信的底层封装）。我们要实现某些功能，肯定会跟各种数据打交道，又或者跟 RPC 之类的东西打交道。 而且完全分离业务逻辑与技术细节是不可能的，因为完全分离也会带来其他的问题，很多简单的场景下还不如写到一块去直观。

只能说，尽量保证方法短小，我们有很多时候可以选择将方法中的一些处理技术细节的代码抽取出去（Extract Method），比如在其中混杂了一个复杂的查询，写了十几行了，这个时候拆分出去或许会更好，让原方法只保留核心业务逻辑的描述。 如果原来方法本身就很短，并且看多两眼就能看懂在做什么的话，拆分了反而会让代码变得更加复杂，这个时候就没必要拆分了。当然，代码的好与坏不应当用这种 “含混的代码美学” 来判断，因为这会带来一个困难是，每个人对于代码理解不一样，所以做出的判断不一样。 但是，这种分歧是没有办法避免的，可能有个办法是，每一个人负责某一功能模块，不要大家改同一块代码，但这个人离职之后，这个人的代码其他人还是得去维护；又或者，另外一个办法是，定一个标准或者说规范，然后大家共同遵守，比如方法不能超过多少行。

总结

可能我们也知道方法长了得拆分一下，因为我们都知道，代码太长看起来太费劲了，但是可能我们并不是很清楚背后的一些原因，导致我们在实践的时候做不到很好地去拆分。 本文从一个新的角度探讨了如何通过分离业务逻辑与技术细节来简化代码，提高可读性和可维护性。合理地对业务逻辑和技术细节进行分离，可以让读者在阅读代码的时候，更容易理解代码背后的业务，而不是被技术细节所迷惑。

对于程序员来说，往往工作几年以后，如果使用的技术没有太大变化，写起代码来往往会有一种得心应手的感觉，好像不需要过多思考就能写出来。 在业务简单的情况下，这样子不会有任何的问题，因为面对的问题足够的简单，根本没有太多发挥的空间。 可是当业务变得复杂起来的时候，这种感觉就会变得非常不靠谱，明明脑子想得通，但是代码写不通，又或者不知道如何下手。

不管是看别人的代码，还是看自己的代码，总会有出现这种感觉的时候。 我们可能会想，明明以前写得那么顺畅，怎么到今天就很难接着原来的代码继续写下去呢？ 可能在实现那个需求的时候，那样实现确实是没问题的，但是那样写出来的代码，可能并不适应新的需求，也就是说，我们的代码并不够灵活。 又或者是，我们忽略了脑海中曾经浮现过的一些比较理性的思考，选择了比较顺畅的、耗时更短的方法，为了赶在 deadline 前完成任务，当然这也是一种选择。

但如果我们想要写出更好维护的代码，那么我们就需要更多地依赖一些理性的思考，而不是感觉。 也就是需要花时间思考一下当前软件的设计是否合理，是否能够适应可遇见的变化，是否能够方便地扩展等。

你真的需要 setter 吗？

当我们想为一个类添加一个属性的时候，我们可能会想到使用 setter 方法来设置这个属性，这是一种非常符合感觉的做法。这样写代码我们也不会有太多的阻力，因为所见即所得。

在实际工作中，我们可能会经常看到这样的代码：

1
2
3
4
5

public void approve(final long bookId) {
  ...
  book.setReviewStatus(ReviewStatus.APPROVED);
  ...
}

这里讨论的不是实体类的 setter

也就是通过 setter 方法来修改对象的属性，这样的代码看起来很简单，但是却隐藏了一些问题：

1. 缺乏封装：在这里，我们本意要实现的操作是 “审核”，但是通过 setter 方法来修改对象的属性，使得我们的操作不再是 “审核”，而是 “修改属性”。
2. 变化不可控：任何人在任何地方都能使用 setter 方法来修改对象的属性，这样会导致对象的状态变得不可控。很难追溯变化是从哪里引起的。

关于第二点，我们可以用数据库的变更来类比一下：

在 Java 中，对数据库的操作，往往有一层 dao 层，这一层负责和数据库交互，而不是直接操作数据库。 这样一来，对数据库的操作就变得可控了，我们可以通过查看 dao 层的方法被哪里调用，来追溯数据库的变更。 相比之下，在某些脚本语言中，比如 PHP，因为书写很自由，没有明显的分层，数据库操作的代码散落在各个地方，导致数据库的变更变得不可控； 代码仓库变得越来越庞大的时候，我们就很难得知某个字段是在哪里被修改的，这样一来，维护成本就会变得非常高，因为操作数据库变更的代码有各种形态，无法简单地通过搜索得到； 这种情况下，我们只能在出问题的时候出修复脚本去修复一些出错的数据，而某段本来应该被修复的代码，埋藏在一个不为人知的角落里。

过多的 setter 透露着一些问题：

1. 对象暴露了太多的细节，缺乏封装。
2. class 的职责不明确/职责过多，如果职责比较单一，那它依赖的参数应该也是比较固定的。
3. 对其他开发者而言，可能不知道什么时候该调用 setter 方法，因为暴露了 setter 之后，意味着 setter 对应的属性是可 set 也可不 set，等于给其他人出了一个难题，因为这个时候他们必须考虑传和不传会有什么影响。

这样的 setter 多了之后，作为维护的人心智负担会越来越重，可能你处理一个业务的时候，就得想很多次该不该 set 某个属性的问题。

不用 setter，那用什么？

好了，理性地思考一番过后，我们知道了 setter 方法的问题，那么我们应该怎么做呢？

这需要分情况讨论，下面是两种非常常见的情况：

1. 更新对象的属性/状态：如上面这个例子。
2. 参数传递：有时候我们在修改一个类的时候发现需要加参数的话，可能有的人会选择加一个 setter 方法来传递新的参数。

针对第一种情况，我们可以简单地封装一个方法，方法名表示我们想要做的操作，比如，上面这个例子，我们的本意是 “审核”：

1
2
3
4
5

public void approve(final long bookId) {
  ...
  book.approve();
  ...
}

在 book 对象中，我们可以定义一个 approve 方法，这个方法负责修改对象的状态：

1
2
3

public void approve() {
  this.reviewStatus = ReviewStatus.APPROVED;
}

也许你会觉得这样做有些多余，但是这样写了之后，我们就知道，任何审核操作都是通过 approve 方法来实现的，如果我们想知道审核状态是哪里修改的，可以直接查看哪里使用了 approve 方法即可。

针对第二种情况，我们有两种办法：

1. 通过构造函数传递参数：如果是全局的参数，也就是整个类都需要用到的参数，我们可以通过构造函数传递参数，然后设置为对象的属性。
2. 通过方法传递参数：如果只是某个方法的参数，我们就不需要在 class 上增加属性，然后通过 setter 方法来传递参数，直接在方法参数中传递即可。

你应该重构那些遗留代码吗？

很多时候，我们在看到某些屎山代码的时候，会有一种冲动，想要将这些代码全部重构一遍，但是这样做真的好吗？或者说，重构后会更好吗？还是会引入更多的问题？ 可能很多人回答不了这几个问题，“屎山” 特征其实很明显，但是真让我们去改的时候，未必就能改得更好。因为我们只是感觉上觉得 “屎”，但是缺乏具体的理性分析。 在这种情况下，就算贸然重构，也还是会存在很多新的问题，可能其他人看了也还是觉得看不懂、不好维护。

这就好比，虽然我们不会做菜，但是我们知道这道菜不好吃，但是我们不知道哪里不好吃，也不知道怎么做才好吃。 如果你让一名美食家来品尝的话，他真的可以给你说出个所以然来，比如他们会针对火候、口感、味道各方面进行分析。 同样的，可能我们不知道怎样写才更好，但是看到那些写得不好的代码的时候，其实我们是有感觉的。 但真要让我们说的话，可能又说不出来。

在这种情况下，我们看到好不好的时候，我们还是可以评判的，但是我们很难往好的方向去改进，因为我们都是凭感觉。

在还没有明确要重构的代码存在哪些问题之前，选择维持现状可能是一个更好的选择。

那我们与 “美食家” 的区别又在哪里呢？

我们知道，我们度量长度的时候，需要有一把尺子，这把尺子就是我们的标准。 同样的，对于美食家而言，他们也有一把尺子，但不同于直尺，他们的尺子是多维度的，比如有火候、口感、味道几个维度等等：

而对于普通食客而言，他们的尺子可能只有一个维度，就是好吃不好吃，也就是本文所说的感觉。

衡量代码的尺子

同样的，对于代码而言，我们也有一把尺子，比如可能我们关注的是以下几个维度（可能每个团队关注的维度不一样，这里这是抛砖引玉）：

1. 可读性：代码是否容易阅读，变量和函数命名是否清晰且具有描述性，注释是否能够帮助理解代码逻辑。
2. 可维护性：代码是否容易维护，是否容易修改；结构是否清晰。
3. 整洁性：代码是否整洁，是否有冗余代码。
4. 可复用性：代码是否可以在其他项目或模块中复用？是否遵循了 DRY（Don't Repeat Yourself）原则？
5. 灵活性：代码是否容易适应变化？是否使用了设计模式等提高代码灵活性的方法？
6. 一致性：代码风格是否一致？是否遵循了团队的代码规范？

评判的维度并不固定，取决于你们更关注哪方面的问题。比如我们这个图就没有把性能这一维度加上去，因为一个项目中，并没有那么多的性能问题，在优化性能之前，把代码写得更好维护可能更重要。

重构的时机

现在，我们知道了，对于代码的好坏，也有一把尺子，那么我们何时应该重构呢？ 我觉得应该是拿 “尺子” 好好量过的时候，也就是，我们从各个维度上去看这段代码，看看它是否符合我们的标准。

在我们没有比较清晰的标准的时候，我们就不要轻易去重构，因为我们不知道重构后的代码是否会更好，还是会引入更多的问题。 有了方向，我们就能知道该怎么做了，同时做完之后，我们也能依据这些标准来判断我们的重构是否成功。

如何摆脱对感觉的依赖？

当然，这里并不是说靠感觉不好，只是说，如果你的感觉并不准的话，我们最好还是有一些可以依赖的标准（代码规范）。

拿一些竞技游戏来说，很多人可能玩很久，依然是个比较菜的水平，他们可能甚至花了 1 万个小时在上面，从 1 万小时理论来说，这些人应该是个专家了，可惜现实并不是这样的，又菜又爱玩的人其实不少，比如我。 经过长时间的玩耍，他们有一些比较模糊的感觉，但是缺乏反馈，所以他们的水平并没有提高；对于这部分玩家来说，游戏只是个消遣，比如我，他们并不会去深究游戏的规则，也不会去看一些攻略，只是单纯地玩而已，这些人就是一年经验用了十年。 而有的人，玩起来就很认真，他们会去看一些攻略，去了解游戏的规则，并且在跟其他玩家对抗的时候，不断改进自己的策略，这样一来，他们的水平才会提高。

好了，我想说的是，如果没有刻意地去思考，我们的感觉都只是一种非常不靠谱的感觉，如果一直依赖于这种感觉，我们也很难做得比昨天好。 专业跟业余的差别就在于，专业的人对一件事总有个 1234，能道出个所以然来，而业余的人只是感觉上觉得这样做好，那样做不好。 结果就是，业余的人偶尔能取得好的成绩，而专业的人可以持续地取得好的成绩。

如果我们想摆脱对感觉的依赖，我们就需要有一些可以依赖的标准，也就是上面所说的 “尺子”。 对于程序员来说，写代码的时候，我们可能有如下标准：

1. 命名规范：变量、函数、类的命名是否符合规范？
2. 是否符合设计原则：比如 SOLID 原则、DRY 原则等。
3. 是否足够整洁：是否有冗余代码？

有很多可以判断代码好坏的标准，我们可以根据这些标准来判断我们的代码是否符合规范，是否可以更好维护。 当我们的这把 “尺子” 越来越好用的时候，我们也就逐渐摆脱了对感觉的依赖，我们分析的时候会更加理性。

当我们觉得某样东西好、不好，但是又说不出所以然来的时候，可以尝试找一些标准来判断。 比如针对产品经理给你的原型、需求，你可能觉得不好，但是你又说不出哪里不好；你可以依据下面的标准来判断一下：

1. 是否有一个好的用户故事（User Story），能让大家都清楚要解决的问题？
2. 清晰性：需求是否明确且易于理解？
3. 完整性：是否考虑了不同的用户场景？
4. 针对背后的问题，当前的产品设计是否就是最好的解决方案？

同样的，如果从来没做过菜的人去买菜，眼花缭乱的也不知道应该怎么选。在经历多次购买之后，他们会形成一些自己的判断标准，比如选菜的时候看新鲜度、颜色、看手感等等。

想摆脱对感觉的依赖，我们就得找到一把合适的 “尺子”，对程序员来说，《代码整洁之道》、《重构》里面包含了一些可以依赖的标准。

什么时候可以依赖感觉？当我们的感觉越来越靠谱的时候，这个时候的感觉就不再是简单的感觉了，而是一种洞见，也就是 “insight”，也就是我们很熟悉各种维度的判断标准的时候。 当然，我们并不可能等到了解了所有的标准之后，才能开始写代码；只是说，不管怎么做，我们总得有一些明确的理由。准确来说，我们并不是要完全摆脱对感觉的依赖，而是要摆脱对不靠谱感觉的依赖。

总结

如果用一句话来总结的话，那就是：不管是写代码还是其他事情，在自己的感觉并不靠谱、不能依赖于感觉的时候，尝试先找一把 “尺子”，也就是找一些可以依赖的判断标准， 熟悉了这些标准之后，我们就可以摆脱对不靠谱感觉的依赖，结果就是，我们的判断会更加准确，我们的决策也会更加理性。

背景

由于业务需要，最近又需要对业务系统中的一块代码做一些修改，加一些新功能。 这块代码做的事情主要是：根据系统设定的一些不同类型的价格，计算得到用户在终端上看到的价格，比如针对不同的客户级别展示不同的价格、又或者针对单个客户设定一个价格等等。

但因为年代久远，年久失修，这块代码已经变得非常复杂，很难维护了，而且原作者已经离职多年。 每每翻看这部分代码都十分头疼，几度想要重构，但又有一些不去重构的理由，比如：

• 影响范围广：用户在终端看到的大多数价格都是通过这块代码计算得到的，一旦出错，影响面很大。
• 逻辑复杂：不过这里指的是代码写得过于复杂了，其实这块业务本身并不复杂，属于比较正常的线性结构，只是流程多了几个步骤。
• 没有办法完全理解原作者写这段代码的思路，主要由于：
  • 代码不规范：变量名、函数名、注释等等都不规范，理解成本偏高。
  • 类的层次结构混乱：没有比较明显的层次结构（比如说自顶向下、逐步细化这样去实现）。
  • 比较分散的业务逻辑数据：这里主要指的是一条数据，被拆分为多个字段，不同字段丢进不同的 hash 表里面，在使用的时候再从不同的 hash map （关联数组）里面取出来。操作起来非常繁琐，也不符合直觉。
  • 充斥了太多奇怪的逻辑，因此也很难保证重构后的代码就是正确的。

鉴于以上原因，一直没有动手去重构这块代码，后续的业务改动只是在原有的基础上做了一些小的修改。 但是很多时候，一些简单的业务逻辑，都需要花费不少时间去理解这块代码，去找到正确的修改点，非常痛苦。

新功能怎么做？

跟往常一样，还是先去找找从什么地方改比较合适，然后再去改。 结果也跟往常一样，在代码里面跳来跳去，然后找到新代码应该待的那个地方。 但是还有令人头疼的地方：

• 新业务所需要的一些依赖的数据，从哪里获取？如何使用？这个问题还好解决，遵循原来的逻辑，可以在 class 里面定义一个字段，在构造函数里面初始化，然后在需要的地方使用。
• 旧的代码里面包含了大量的 hash map（也就是关联数组），大部分时候，用起来还好。但是，导致的结果是，后续维护的时候，如果不写清楚注释，无法得知这个变量里面装了什么字段，只能通过调试的方式打印出来。这样的代码，不仅不好维护，而且也不好理解。

当然，我们也可以选择忍着头疼，继续往下写，但是如果新加入的代码出 bug 了，我们也得忍着头疼去找 bug，显然不是一个好的选择。

为什么选择重构？

对于这块遗留代码（legacy code），为了后续不那么头疼，最终选择了重构。当然也不是一时兴起，主要是因为时机成熟了：

• 修改旧代码的成本已经高过重构然后再修改的成本了（包括理解旧代码的成本、修改成本、后续维护的成本）。
• 这块业务本身并不复杂，只是代码写得复杂了，重构后会更好维护，会有更加清晰的结构。就算出现 bug，也能快速定位以及修复。
• 这一两年，对于如何写好这类代码有了一些新的认识，可以尝试将这些新的认识应用到这块代码中（此前很多次有过重构的念头，但是一直都没有什么好的办法）。

如何重构及为何

原代码有太多可圈可点的地方，但是本文只讲述个人觉得最重要的地方（嗯，也就是 "主要矛盾"）：

1. 如何得到一个良好的抽象（可以简单地理解为类），以及这个抽象应该包含什么属性、能力（方法）？
2. 如何组织这些抽象概念，使得代码更加清晰？

个人以为，处理好了这两个问题，大概率可以得到相对比较好维护的代码。

什么是抽象？

在开始这一小节之前，有必要讲一下抽象。关于抽象的定义，百度百科中是这样说的：抽象是从众多的事物中抽取出共同的、本质性的特征，而舍弃其非本质的特征的过程。

这么说可能有些抽象，举个不那么恰当的例子，我们可以对比一下下面两个图：

第一个图是一个具象的图，经常会有人拿这个表情来作为调侃。然后当我们见多了之后，有人直接发图二了，虽然上面只有几根线条，但是在特定场景下，我们看这几根线条就能知道对方想表达什么了。

从图一到图二的过程，我们可以理解为抽象的过程。这个过程中，抽取了其中最关键的特征，舍弃了其他的特征。

在实际的开发中，其实我们代码中的那些对象也都是一个个抽象出来的概念，不可能涵盖一个真实对象的所有特性，比如 Sku 对象，用来表示商品的时候，我们只会给它加上我们实际业务所需要的那些特性，比如颜色、价格等等。

定义虽然这么说，但落实到代码中的时候，并不需要抽取那么多共同的、本质性的特征，我们只需要业务处理所需要的那些特征。

又或者，在我们数据库模型的基础上，根据实际业务引申出一些新的抽象，比如员工，可以引申出经理、普通员工等等。这其实是在原有抽象的基础上，添加一些新的特性，从而产生新的抽象。 在实际开发中，可能后者对我们来说作用更大，很多时候我们都是在处理各种模型之间的关系，而不是处理单一的对象。

我们的数据库模型是根据业务流程所需要的各种数据根据其不同种类、属性、特性拆分开了的，为了更好地处理我们的业务逻辑，我们可以在业务代码中再将一些有关联的数据组织起来，形成一个新的抽象，从而更好地实现复用。

如何得到一个良好的抽象？

我们可能习惯了将所有的业务逻辑丢到 service 里面去实现，似乎 service 是个万能的东西，这样造成的结果是，我们会丢失很多业务上的语义，丢失业务场景， 带来最大的麻烦是，后续维护的人理解起来会非常费劲，需要通过人脑将那些割裂的信息重新组织起来，这是一个非常费劲的过程，往往会造成一些理解上的偏差。 当然，一些简单的 CURD 并不在本文的讨论范围之内。

面向对象分析（OOA）告诉我们，我们可以通过分析业务逻辑，找到其中的对象，然后将这些对象抽象出来，形成一个类，通过这个步骤，我们得到了数据库表、ORM 模型。 在此基础上，针对不同的业务场景，我们也可以衍生出不同的类（这是本文主要讲述的东西），一个更加契合我们实际业务场景的类。

简单来说，我们可以通过下面这几个步骤来得到这么的一个类：

1. 为新的类命名：命名的规则可以是 业务场景 + 实际要操作的对象。
2. 为这个新的类加上需要处理的业务场景所必须的属性。
3. 为这个新的类加上需要处理的业务场景所必须的方法。

下面举一个比较实际的例子，比如，假设我们需要组合一些信息来计算用户在商城实际购买商品的价格，我们有如下几个模型：

1. Sku：包含商品本身的的一些基本信息。
2. User：包含用户的一些基本信息，其中包含了用户的等级。
3. SkuLevelPrice：记录了 Sku 对应的不同等级的价格。
4. SkuUserPrice：记录了针对单个用户设置的价格。

我们遵循一下上面说到的三个步骤，实践一下：

1. 命名：PriceSku。我们的业务场景是计算价格，操作额度对象是 Sku，这个命名也许不太好，但 “又不是不能用”。不管怎样，起码反映了业务和要操作的对象，不算太糟糕，另外主要是，我也没有想到更好的命名（但是很多时候，对于命名的这种考量是非常有意义的）。
2. 为 PriceSku 添加几个属性：Sku、User、SkuLevelPrice、SkuUserPrice。这些属性是我们计算价格所必须的，也是我们的业务场景所必须的。
3. 为 PriceSku 添加一个个方法：getPrice。这个方法的作用是计算用户在商城实际购买商品的价格。

最终的伪代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class PriceSku:
    // 属性定义
    sku: Sku                      // 商品对象
    user: User                    // 用户对象
    skuLevelPrice: SkuLevelPrice  // 商品等级价格对象
    skuUserPrice: SkuUserPrice    // 用户特定价格对象

    // 构造函数
    constructor(sku, user, skuLevelPrice, skuUserPrice):
        this.sku = sku
        this.user = user
        this.skuLevelPrice = skuLevelPrice
        this.skuUserPrice = skuUserPrice

    // 方法定义
    method getPrice():
        // 检查是否有针对单个用户的价格
        if this.skuUserPrice.isPriceSetForUser(this.user.id):
            userPrice = this.skuUserPrice.getPriceForUser(this.user.id)
            return userPrice  // 返回用户特定价格

        // 如果没有用户特定价格，则根据用户等级获取价格
        if this.skuLevelPrice.hasPriceForLevel(this.user.level):
            return this.skuLevelPrice.getPriceForLevel(this.user.level)  // 返回等级价格

        // 如果都没有，返回默认价格（可以根据需求定义）
        return this.sku.getDefaultPrice()  // 返回默认价格

在这个例子中，我们将计算价格的逻辑抽象出来，形成了一个新的类 PriceSku，这个类包含了我们计算价格所必须的属性和方法，这样我们就可以在其他地方直接使用这个类，而不用再去关心这个类的内部实现。 同时，如果我们需要修改计算价格的逻辑，只需要修改这个类的 getPrice 方法即可，在主流程中加一个 if-else 语句。

看到这个，可能有人会条件反射般地想到这可以用策略模式（有不少文章经常吹嘘要消灭 if-else，各种标题党）。确实，这种场景是可以用策略模式的，但这应该发生在 PriceSku 变得越来越庞大的时候，而不是一开始就用策略模式。 一上来就用策略模式，可能会让代码变得更加复杂，不利于维护。如果我们担心代码在后续业务复杂之后变得难以维护，另外一个选择是想办法让目前的代码写得更简洁一点，更容易理解一点， 而不是在业务变得复杂之前，先把代码写复杂了，那样就南辕北辙了。

如何组织这些抽象概念？

通过上一步，我们得到了一个可用的类了，但是在我们开始使用的时候，我们可能会发现，好像还是不太好用。 因为在实际的业务场景中，我们很多时候都是需要做一些批量的查询，比如，用户查看一个商品列表的时候，我们需要查询一组商品的价格，而不只是一个商品的价格。

在这种场景下，我们能想到的一个很简单的办法是，写一个循环来逐个获取，这样也是可行的。只是存在一个问题是，如果很多地方都需要做批量的处理，会有比较多冗余的代码。 但是这对我们来说应该毫无难度，我们可能会选择在某个地方定义一个批量查询的方法，不管是什么地方，总会有一个地方，service 也好、controller 也好。

然后某一天，有一个新的需求，也是跟价格相关的，我们需要改造一下 PriceSku，让它只返回某个商品的会员等级价，同样的，我们也需要做批量的查询。 然后也难不住我们，在 PriceSku 里面添加一个获取会员等级价的方法，然后在某个地方再写一个批量查询的方法。到目前为止还没有什么问题。

但是问题还是存在的，我们似乎缺少了一个地方，这个地方作为承载所有批量操作 PriceSku 的地方。当然放 service 也能跑，又不是不能用，只是最终结果是 service 越来越膨胀，承载的职责越来越多。 在这种情况下，我们可以考虑定义一个 PriceSkuCollection 类，在里面做那些批量操作，比如下面这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

class PriceSkuCollection:
    // 属性定义
    priceSkus: List<PriceSku>  // 价格对象列表

    // 构造函数
    constructor(priceSkus):
        this.priceSkus = priceSkus

    // 方法定义
    method getPriceForAll():
        prices = []
        for priceSku in this.priceSkus:
            prices.append(priceSku.getPrice())
        return prices
    
    // 只获取会员等级价
    method getLevelPriceForAll():
        levelPrices = []
        for priceSku in this.priceSkus:
            levelPrices.append(priceSku.getLevelPrice())
        return levelPrices

通过这种方式，我们可以有如下几个好处：

1. 内聚性更强：PriceSkuCollection 只负责批量操作 PriceSku，不会有其他的职责。
2. 代码更加清晰：PriceSkuCollection 里面的方法都是批量操作 PriceSku 的方法。
3. 同时可以在此基础上定义一些获取元数据的方法，当然，在我们这个例子中，并没有什么元数据可供我们获取。但是如果某一组 PriceSku 本身是一组有实际业务意义的数据，那么我们这种定义方式就有非常大的优势。

比如，如果我们的一组 PriceSku 代表了用户加入到采购车的所有商品，那么我们可以在 PriceSkuCollection 里面定义一个 getTotalPrice 方法，用来计算这一组 PriceSku 的总价格。 当然，在实际业务场景中，我们需要获取的元数据不仅仅是价格，还有很多其他的数据，比如数量等等，这时候我们就可以在 PriceSkuCollection 里面定义一些方法来获取这些数据。 这个时候，PriceSkuCollection 就变成了一个非常实用的类，可以用来处理一组 PriceSku 的数据。

好了，说了这么多，我想表达的是，在建立起对业务的抽象的时候，我们还得考虑一下抽象的层次性，本文的例子中，我们通过 PriceSku 和 PriceSkuCollection 两个类，将业务抽象分成了两个层次，这样可以更好地组织我们的代码。 其中，PriceSku 用来处理单个 Sku 的价格计算，PriceSkuCollection 用来处理一组 Sku 的价格计算，职责非常明确，代码也更加清晰。

有时候业务逻辑可能并没有那么复杂，但是因为抽象的层次混乱、或者完全没有抽象的层次，各种不同抽象层次的代码混杂在一起，从而使得业务看起来非常复杂。 业务的复杂是不可避免的，任何一个项目只要在发展，都会越来越复杂，合理组织代码，将代码分成不同的抽象层次，是我们应对复杂业务的一个重要手段。

网络通信已经非常复杂了，但是网络通信的最底层也不过是网络中传输的二进制数据，但是很多年以前就形成了 TCP/IP 协议，这个协议就是一个非常好的抽象，将网络通信分成了不同的层次，每一层都有自己的职责， 每一层只处理一件事情，下层为上层提供服务，上层调用下层的服务，这样就形成了一个网络通信的协议栈。 而 HTTP 是对 TCP/IP 协议的进一步抽象，将网络通信分成了请求和响应，有了 HTTP，我们开发网络应用就不用自己去处理 TCP/IP 协议了，这样我们就可以更加专注于业务逻辑的处理，而不用去处理网络通信的细节。

同样的，识别出我们业务逻辑中的抽象，然后将这些抽象分成不同的层次，是我们应对复杂业务的一个重要手段。时间长了，新的业务开发的时候，成本更低，因为不再需要理解那些没有抽象的代码，我们需要处理的是一个个具有具体业务意义的抽象。

总结

本文前面的三分之一是在讲述很多人都会面临的一些困境，如果你也有类似的困境，也许可以参考一下本文的一些思路。 总的来说，在基本的数据库表的 ORM 模型基础上，我们可以做更多的抽象，我们可以对需要复用的业务逻辑进行抽象，然后将这些抽象分成不同的层次， 在不同层次的抽象中，处理其对应层次的业务逻辑，然后为更上一层抽象提供服务。