Disruptor在撮合引擎的实践｜得物技术

一

Disruptor的简介

Disruptor的使用场景

Disruptor和ArrayBlockingQueue性能对比

自定义事件和事件工厂

定义事件处理器，即消费者

定义事件生产者

编写启动类

 Disruptor消费者等待策略

单消费者

多消费者：并行的、广播模式

//注册多个消费者disruptor.handleEventsWith(new LongEventHandler(), new LongEventHandler1(), new LongEventHandler2());

多消费者：并行的、消费者组模式

public class LongWorkHandler  implements WorkHandler<LongEvent> {    @Override    public void onEvent(LongEvent longEvent) throws Exception {        System.out.println("Event: " + logEvent);    }}public class OtherWorkHandler  implements WorkHandler<LongEvent> {    @Override    public void onEvent(LongEvent longEvent) throws Exception {        System.out.println("Event: " + logEvent);    }}
//注册消费者组disruptor.handleEventsWithWorkerPool(new LongWorkHandler()        , new LongWorkHandler()        , new LongWorkHandler());

 核心概念

Sequence

Sequencer

SequenceBarrier（序列屏障）

无锁化设计

Disruptor是如何解决伪共享的

预分配内存

C端核心界面

以上截图仅用于技术展示，不构成投资建议

交易系统简化交互图

撮合应用的特点

撮合引擎流程图

撮合引擎之Disruptor代码

定义事件：用户交易单

定义事件处理器：对用户买单和卖单进行撮合匹配

事件生产者：构建Disruptor、生产事件

/** * disruptor启动入口类，当系统读取到当前机器需要为哪些交易对提供服务的时候， * 我们需要为这些交易对进行分组服务，哪些交易对放到同一个disruptor中 * 通过分组，一方面确保了活跃度高的交易对能够最大程度的利用资源，另一方面活跃度低的交易对能够有效处理， * 同时降低了cpu暴涨的风险 */@Service@Slf4jpublic class ExchangeLauncher {
    private static int BUFFER_SIZE = 1024 * 16;    @Resource    private LimitBuyMatchService limitBuyMatchService;    @Resource    private LimitSellMatchService limitSellMatchService;    @Resource    private MarketBuyMatchService marketBuyMatchService;    @Resource    private MarketSellMatchService marketSellMatchService;    @Resource    private MatchClusterConfiguration matchClusterConfiguration;    @Value("${match.worker-queue-size:5}")    private int workSize;    //一个交易对对应一个disruptor    private Map<Integer, ExchangeCore> exchangeCoreMap = new ConcurrentHashMap<>();
    private List<ExchangeCore> exchangeCoreList = new CopyOnWriteArrayList<>();
    public void start() {        try {            //init order book            OrderBook.init();
            Set<Integer> symbolIdListSet = matchClusterConfiguration.getMasterSymbolIdSet();            if (CollectionUtils.isNotEmpty(symbolIdListSet)) {                List<Integer> allSymbolIds = new ArrayList<>(symbolIdListSet);                List<List<Integer>> pageList = ListUtils.partition(allSymbolIds, workSize);                pageList.forEach(symbolIds -> {                    ResultsHandler handler = new ResultsHandler(new HashSet<>(symbolIds), workSize);                    ExchangeCore exchangeCore = new ExchangeCore(handler, BUFFER_SIZE, new NamedThreadFactory("match", false));                    exchangeCore.start();                    exchangeCoreList.add(exchangeCore);                    symbolIds.forEach(symbolId -> exchangeCoreMap.put(symbolId, exchangeCore));                });            }            // 注册matchService子类            registerMatchServices();        } catch (Exception e) {            log.error("exchangeLauncher start error:{}", e.getMessage(), e);        }    }    private void registerMatchServices() {        MatchStrategy.register(OrderTypeEnum.LIMIT, SideEnum.BUY, limitBuyMatchService);        MatchStrategy.register(OrderTypeEnum.LIMIT, SideEnum.SELL, limitSellMatchService);        MatchStrategy.register(OrderTypeEnum.MARKET, SideEnum.BUY, marketBuyMatchService);        MatchStrategy.register(OrderTypeEnum.MARKET, SideEnum.SELL, marketSellMatchService);    }  }

public class ExchangeCore extends AbstractLifeCycle {
    private final Disruptor<DisruptorEvent> disruptor;
    private MatchEventPublisher publisher;
    private ResultsHandler eventHandler;
    public ExchangeCore(ResultsHandler matchHandler, int ringBufferSize, ThreadFactory threadFactory) {        EventFactory eventFactory = () -> new DisruptorEvent();        this.disruptor = new Disruptor<>(eventFactory, ringBufferSize, threadFactory);        publisher = new MatchEventPublisher(this.disruptor);        disruptor.setDefaultExceptionHandler(new DisruptorExceptionHandler());        this.eventHandler = matchHandler;        disruptor.handleEventsWith(eventHandler);        disruptor.start();    }
    @Override    public void start() {        super.start();    }
    @Override    public void stop() {        super.stop();        disruptor.shutdown();    }

    public BaseResponse doMatch(EntrustOrder taker) {        // 前置处理----start        if (OrderTypeEnum.getOrderTypeEnum(taker.getOrderType()) == null || SideEnum.getSideEnum(taker.getSide()) == null) {            log.error("{} - parameter error:{} or {}", taker.getTraceId(), "orderType", "side");            return BaseResponse.error(TradingMatchCodeEnum.PARAMETER_ERROR);        }
        MatchStatusEnum matchStatusEnum = OrderBook.getByOrderId(taker.getOrderId());
        MetricService metricService = SpringContextUtil.getBean(MetricService.class);        MatchClusterConfiguration configuration = SpringContextUtil.getBean(MatchClusterConfiguration.class);        // 撮合防重校验，并发存在问题。但是消费的时候，是单线程，做了校验，不存在重复撮合的问题。        if (OperationTypeEnum.MATCH.getCode() == taker.getOperationType()) {            if (matchStatusEnum != null) {                //短时间内重复撮合                log.error("{} - match repeat ,orderId :{}", taker.getTraceId(), taker.getOrderId());                return BaseResponse.error(TradingMatchCodeEnum.REPEAT_REQUEST);            }            //构造对象进入等待队列            OrderBook.addToOrderBook(taker.getOrderId(), MatchStatusEnum.WAIT_ING);            metricService.count(MetricNames.ORDER_TYPE_NUM, "type", "match", "group", configuration.getClusterName());        } else if (OperationTypeEnum.CANCEL.getCode() == taker.getOperationType()) {            int cancelType = taker.getCancelType();            /**             异常单-产生情况：收单服务 调用撮合 出现异常，不知道成功没，没有明确响应 开始进行异常撤单             */            if (CancelTypeEnum.NORMAL_CANCEL.getCode() == cancelType) {                if (matchStatusEnum == null) {                    // 数据有可能在请求队列中被逐出，需要继续走逻辑                    //                } else {                    if (MatchStatusEnum.MATCH_END == matchStatusEnum) {                        //重复撤销，深度盘已经没有数据，没必要继续往下,不走disruptor 和撮合直接返回                        log.error("{} - cancel failed, match end ,orderId :{}", taker.getTraceId(), taker.getOrderId());                        return BaseResponse.error(TradingMatchCodeEnum.REPEAT_REQUEST);                    }                    OrderBook.addToOrderBook(taker.getOrderId(), MatchStatusEnum.WAIT_CANCEL);                }            } else {                // reload异常撤单，要加入内存                OrderBook.addToOrderBook(taker.getOrderId(), MatchStatusEnum.WAIT_CANCEL);            }        } else {            log.warn("--------can not find the operationType[{}]", taker.getOperationType());            throw new TradingMatchException("can not find the operationType[" + taker.getOperationType() + "]");        }        // 前置处理----end        //Disruptor开始发布事件        publisher.publish(taker);        return BaseResponse.success();    }
    public Disruptor<DisruptorEvent> getDisruptor() {        return disruptor;    }}

public class MatchEventPublisher {    private Disruptor<DisruptorEvent> disruptor;
    public MatchEventPublisher(Disruptor<DisruptorEvent> disruptor) {        this.disruptor = disruptor;    }
    private static final EventTranslatorOneArg<DisruptorEvent, EntrustOrder> TRANSLATOR =            (event, sequence, entrustOrder) -> {                event.setEntrustOrder(entrustOrder);            };
    public void publish(EntrustOrder taker) {        RingBuffer<DisruptorEvent> ringBuffer = disruptor.getRingBuffer();        taker.setSequence(ringBuffer.getCursor());        taker.setArriveTime(System.currentTimeMillis());        ringBuffer.publishEvent(TRANSLATOR, taker);        // ...    }
}

五

总结