Spring Cloud开发 - 设计定序系统 - 《廖雪峰 Java 教程（Java 20）》

设计定序系统
如何在定序器重启后正确初始化下一个序列号？
如何在定序器崩溃后自动恢复？
如何解决定序的性能瓶颈？
参考源码
小结
读后有收获可以支付宝请作者喝咖啡：

设计定序系统

当系统通过API接收到所有交易员发送的订单请求后，就需要按接收顺序对订单请求进行定序。

定序的目的是在系统内部完成订单请求排序，排序的同时给每个订单请求一个全局唯一递增的序列号，然后将排序后的订单请求发送至交易引擎。

因此，定序系统的输入是上游发送的事件消息，输出是定序后的带Sequence ID的事件，这样，下游的交易引擎就可以由确定性的事件进行驱动。

除了对订单请求进行定序，定序系统还需要对撤消订单、转账请求进行定序，因此，输入的事件消息包括：

OrderRequestEvent：订单请求；
OrderCancelEvent：订单取消；
TransferEvent：转账请求。

对于某些类型的事件，例如转账请求，它必须被处理一次且仅处理一次。而消息系统本质上也是一个分布式网络应用程序，它的内部也有缓存、重试等机制。一般来说，消息系统可以实现的消息传输模式有：

消息保证至少发送成功一次，也就是可能会重复发送（At least once）；
消息只保证最多发送一次，也就是要么成功，要么失败（At most once）；
消息保证发送成功且仅发送成功一次（Exactly once）。

实际上，第3种理想情况基本不存在，没有任何基于网络的消息系统能实现这种模式，所以，大部分消息系统都是按照第1种方式来设计，也就是基于确认+重试的机制保证消息可靠到达。

而定序系统要处理的事件消息，例如转账请求，如果消息重复了多次，就会造成重复转账，所以，我们还需要对某些事件消息作特殊处理，让发送消息的客户端给这个事件消息添加一个全局唯一ID，定序系统根据全局唯一ID去重，而不是依赖消息中间件的能力。

此外，为了让下游系统，也就是交易引擎能一个不漏地按顺序接收定序后的事件消息，我们也不能相信消息中间件总是在理想状态下工作。

除了给每个事件消息设置一个唯一递增ID外，定序系统还同时给每个事件消息附带前一事件的ID，这样就形成了一个微型“区块链”：

┌─────┐   ┌─────┐   ┌─────┐   ┌─────┐
│sid=1│   │sid=2│   │sid=3│   │sid=4│
│pid=0│──▶│pid=1│──▶│pid=2│──▶│pid=3│
│msg=A│   │msg=B│   │msg=C│   │msg=D│
└─────┘   └─────┘   └─────┘   └─────┘

由于下游接收方可以根据Sequence ID去重，因此，重复发送的消息会被忽略：

┌─────┐┌─────┐┌─────┐┌ ─ ─ ┐┌ ─ ─ ┐┌─────┐
│sid=1││sid=2││sid=3│ sid=2  sid=3 │sid=4│
│pid=0││pid=1││pid=2││pid=1││pid=2││pid=3│
│msg=A││msg=B││msg=C│ msg=B  msg=C │msg=D│
└─────┘└─────┘└─────┘└ ─ ─ ┘└ ─ ─ ┘└─────┘

如果出现消息丢失：

┌─────┐┌─────┐┌ ─ ─ ┐┌─────┐
│sid=1││sid=2│       │sid=4│
│pid=0││pid=1││     ││pid=3│
│msg=A││msg=B│       │msg=D│
└─────┘└─────┘└ ─ ─ ┘└─────┘

由于存在Previous ID，下游接收方可以检测到丢失，于是，接收方可以根据上次收到的ID去数据库查询，直到读取到最新的Sequence ID为止。只要定序系统先将定序后的事件消息落库，再发送给下游，就可以保证无论是消息重复还是丢失，接收方都可以正确处理：

┌─────────┐   ┌─────────┐   ┌─────────┐
│Sequencer│──▶│   MQ    │──▶│ Engine  │
└─────────┘   └─────────┘   └─────────┘
     │        ┌─────────┐        │
     └───────▶│   DB    │◀───────┘
              └─────────┘

整个过程中，丢失极少量消息不会对系统的可用性造成影响，这样就极大地减少了系统的运维成本和线上排错成本。

最后，无论是接收方还是发送方，为了提高消息收发的效率，应该总是使用批处理方式。定序系统采用批量读+批量batch写入数据库+批量发送消息的模式，可以显著提高TPS。

下面我们一步一步地实现定序系统。

首先定义要接收的事件消息，它包含一个Sequence ID、上一个Sequence ID以及一个可选的用于去重的全局唯一ID：

public class AbstractEvent extends AbstractMessage {
    // 定序后的Sequence ID:
    public long sequenceId;
    // 定序后的Previous Sequence ID:
    public long previousId;
    // 可选的全局唯一标识:
    @Nullable
    public String uniqueId;
}

定序系统接收的事件仅包含可选的uniqueId，忽略sequenceId和previousId。定序完成后，把sequenceId和previousId设置好，再发送给下游。

SequenceService用于接收上游消息、定序、发送消息给下游：

@Component
public class SequenceService {
    @Autowired
    SequenceHandler sequenceHandler;
    // 全局唯一递增ID:
    private AtomicLong sequence;
    // 接收消息并定序再发送:
    synchronized void processMessages(List<AbstractEvent> messages) {
        // 定序后的事件消息:
        List<AbstractEvent> sequenced = null;
        try {
            // 定序:
            sequenced = this.sequenceHandler.sequenceMessages(this.messageTypes, this.sequence, messages);
        } catch (Throwable e) {
            // 定序出错时进程退出:
            logger.error("exception when do sequence", e);
            System.exit(1);
            throw new Error(e);
        }
        // 发送定序后的消息:
        sendMessages(sequenced);
    }
}

SequenceHandler是真正写入Sequence ID并落库的：

@Component
@Transactional(rollbackFor = Throwable.class)
public class SequenceHandler {
    public List<AbstractEvent> sequenceMessages(MessageTypes messageTypes, AtomicLong sequence, List<AbstractEvent> messages) throws Exception {
        // 利用UniqueEventEntity去重:
        List<UniqueEventEntity> uniques = null;
        Set<String> uniqueKeys = null;
        List<AbstractEvent> sequencedMessages = new ArrayList<>(messages.size());
        List<EventEntity> events = new ArrayList<>(messages.size());
        for (AbstractEvent message : messages) {
            UniqueEventEntity unique = null;
            final String uniqueId = message.uniqueId;
            // 在数据库中查找uniqueId检查是否已存在:
            if (uniqueId != null) {
                if ((uniqueKeys != null && uniqueKeys.contains(uniqueId))
                        || db.fetch(UniqueEventEntity.class, uniqueId) != null) {
                    // 忽略已处理的重复消息:
                    logger.warn("ignore processed unique message: {}", message);
                    continue;
                }
                unique = new UniqueEventEntity();
                unique.uniqueId = uniqueId;
                if (uniques == null) {
                    uniques = new ArrayList<>();
                }
                uniques.add(unique);
                if (uniqueKeys == null) {
                    uniqueKeys = new HashSet<>();
                }
                uniqueKeys.add(uniqueId);
            }
            // 上次定序ID:
            long previousId = sequence.get();
            // 本次定序ID:
            long currentId = sequence.incrementAndGet();
            // 先设置message的sequenceId / previouseId，再序列化并落库:
            message.sequenceId = currentId;
            message.previousId = previousId;
            // 如果此消息关联了UniqueEvent，给UniqueEvent加上相同的sequenceId：
            if (unique != null) {
                unique.sequenceId = message.sequenceId;
            }
            // 准备写入数据库的Event:
            EventEntity event = new EventEntity();
            event.previousId = previousId;
            event.sequenceId = currentId;
            event.data = messageTypes.serialize(message);
            events.add(event);
            // 添加到结果集:
            sequencedMessages.add(message);
        }
        // 落库:
        if (uniques != null) {
            db.insert(uniques);
        }
        db.insert(events);
        // 返回定序后的消息:
        return sequencedMessages;
    }
}