所使用的技術(shù)

springcloud,redis,rocketMq,websocket,mysql

架構(gòu)圖

行情推送架構(gòu)圖.png

整體流程說明：交易服務(wù)交易產(chǎn)生的行情辈讶，存放到redis的隊(duì)列內(nèi)命浴，然后行情服務(wù)會(huì)監(jiān)聽這個(gè)隊(duì)列，獲取這個(gè)隊(duì)列里面的行情節(jié)點(diǎn)贱除，經(jīng)過數(shù)據(jù)處理生闲，如，生成k線數(shù)據(jù)月幌，然后存入redis內(nèi)碍讯，并且使用mq異步持久化行情數(shù)據(jù)，同時(shí)扯躺，將行情數(shù)據(jù)發(fā)布給網(wǎng)關(guān)捉兴，網(wǎng)關(guān)訂閱到行情數(shù)據(jù)后蝎困，使用websocket實(shí)時(shí)推送給前端用戶。

問題

1.交易產(chǎn)生的行情數(shù)據(jù)是怎么樣的倍啥？
這里要說一下禾乘，國際行情不是我們?nèi)ギa(chǎn)生的，而是要去對(duì)接第三方（比如對(duì)接OTC）虽缕，拿到行情始藕，然后我們?cè)賹?duì)國際行情進(jìn)行處理推送展示。而本文的行情是我們自己系統(tǒng)自己交易產(chǎn)生的氮趋，所以會(huì)有一個(gè)交易服務(wù)鳄虱，給用戶進(jìn)行交易，然后產(chǎn)生行情凭峡。
產(chǎn)生的行情數(shù)據(jù)其實(shí)很簡(jiǎn)單拙已，大部分都包含：成交價(jià)格，成交量摧冀，成交時(shí)間這三個(gè)字段倍踪，因?yàn)槭亲约寒a(chǎn)生的行情，所以可以把成交的訂單號(hào)也一并放到隊(duì)列里索昂，方便查找行情的出處建车。

2.什么是分時(shí)線，什么是K線椒惨？
分時(shí)線：每分鐘的最后一筆成交價(jià)的連線叫分時(shí)線缤至。分時(shí)線即大盤、個(gè)股分時(shí)走勢(shì)圖中的白色曲線康谆，它反映的是大盤领斥、個(gè)股的實(shí)時(shí)走勢(shì)。像下圖沃暗，橫坐標(biāo)的時(shí)間月洛，縱坐標(biāo)是成交價(jià)格，那么每一分鐘的最后一筆成交價(jià)就的連線就是分時(shí)線了孽锥，分時(shí)線是實(shí)時(shí)變動(dòng)的嚼黔，這里就要使用ws推送行情了。那么這里有幾個(gè)問題惜辑，就是怎么獲取分時(shí)線數(shù)據(jù)唬涧？分時(shí)線數(shù)據(jù)如何存儲(chǔ)？你想盛撑，要產(chǎn)生和存儲(chǔ)分時(shí)圖的數(shù)據(jù)碎节，我們是不是得將每一分鐘的最后一口價(jià)進(jìn)行存儲(chǔ)，并且存儲(chǔ)時(shí)得注明這個(gè)價(jià)格是哪一分鐘的價(jià)格撵彻，這個(gè)要如何實(shí)現(xiàn)钓株？問題會(huì)在后面講到解決辦法实牡。

行情分時(shí)圖.png

K線圖：股市及期貨市場(chǎng)中的K線圖的畫法包含四個(gè)數(shù)據(jù)，即開盤價(jià)轴合、最高價(jià)创坞、最低價(jià)、收盤價(jià)受葛，所有的k線都是圍繞這四個(gè)數(shù)據(jù)展開题涨，反映大勢(shì)的狀況和價(jià)格信息。如果把每日的K線圖放在一張紙上总滩，就能得到日K線圖纲堵，同樣也可畫出周K線圖、月K線圖闰渔。
從圖片上來看席函，其實(shí)不止開盤價(jià)、最高價(jià)冈涧、最低價(jià)茂附、收盤價(jià)這個(gè)數(shù)據(jù)，還有成交量和成交額督弓。
所以营曼，包括的數(shù)據(jù)就有：開盤價(jià)、最高價(jià)愚隧、最低價(jià)蒂阱、收盤價(jià)，成交量狂塘、成交額
比如今天是2020-03-01录煤，那么今天就會(huì)產(chǎn)生一個(gè)日K數(shù)據(jù)，數(shù)據(jù)包含：日期（精確到日睹耐，即2020-03-01）辐赞，開盤價(jià)（今天的開始價(jià)格）、最高價(jià)（今天的最高價(jià)格）硝训、最低價(jià)（今天的最低價(jià)格）、收盤價(jià)（今天的最后一口價(jià)）新思，成交量（今天的總成交量）窖梁，成交額（今天的總成交額）同理，如果是周K夹囚，則就是以周為單位：日期（這周的結(jié)束日期纵刘，即這周的最后一天的日期，如今天周日荸哟，則這周最后一天就是今天2020-03-01）假哎，開盤價(jià)（這周的開始價(jià)格）瞬捕、最高價(jià)（這周的最高價(jià)格）、最低價(jià)（這周的最低價(jià)格）舵抹、收盤價(jià)（這周的最后一口價(jià)）成交量（這周的總成交量）肪虎，成交額（這周的總成交額）。
當(dāng)然惧蛹，這里的日期怎么顯示扇救，要看公司具體需求，比如周K的日期香嗓，有些公司是拿這周的開始日期迅腔，有些公司是拿這周的結(jié)束日期。

行情k線圖.png

3.為什么要將行情保存到redis的隊(duì)列里靠娱，然后行情服務(wù)去主動(dòng)獲取隊(duì)列里的行情沧烈？而不直接使用redis的發(fā)布訂閱，將行情發(fā)布給行情服務(wù)像云；或者使用rocketMq锌雀，將行情異步發(fā)送給行情服務(wù)？
可能有人會(huì)說苫费，可以直接使用redis的發(fā)布訂閱來將行情發(fā)布給行情服務(wù)汤锨，所以就沒必要將行情存放到redis的一個(gè)隊(duì)列里，然后行情服務(wù)再主動(dòng)去獲取行情數(shù)據(jù)百框。理論上闲礼，這樣也是可以達(dá)到效果的，但是為什么不這樣做铐维？主要就是因?yàn)閞edis的發(fā)布訂閱柬泽，是消息不可靠的，也就是說這個(gè)消息很可能丟失嫁蛇，假如行情服務(wù)全部掛掉了锨并，但是交易服務(wù)還是正常的發(fā)布消息蛾号，那么骨杂，這個(gè)時(shí)候發(fā)布的消息將全部丟失暂吉。所以才需要將行情數(shù)據(jù)先保存到一個(gè)隊(duì)列內(nèi)弥搞，然后等待行情服務(wù)主動(dòng)去獲取埂软，這樣么介，即使行情服務(wù)掛了拧咳，行情數(shù)據(jù)也不會(huì)丟失诞挨，等到行情服務(wù)正常后底靠，行情服務(wù)會(huì)主動(dòng)去獲取隊(duì)列里的行情數(shù)據(jù)害晦，一個(gè)一個(gè)的處理。
那么暑中，為什么不直接使用rocketMq將行情發(fā)送給行情服務(wù)呢壹瘟？其中最大的原因就是消息順序消費(fèi)的問題鲫剿，行情需要保證一個(gè)正確的順序輸出的，比如我這一秒產(chǎn)生的行情稻轨，不能夠比上一秒產(chǎn)生的行情要晚推送給前端灵莲，是需要保證順序的，而rocketMq是很難保證順序消費(fèi)的澄者，所以也不會(huì)使用rocketmq來實(shí)現(xiàn)這一點(diǎn)笆呆。

4.websocket是什么？為什么使用它做行情的實(shí)時(shí)推送粱挡？
可能有人會(huì)想赠幕，要保證頁面實(shí)時(shí)的更新獲取最新價(jià)格，可以使用http的長(zhǎng)輪詢询筏，也就是不斷的去輪詢請(qǐng)求接口獲取相應(yīng)的數(shù)據(jù)來保證實(shí)時(shí)性榕堰，但是這樣是很耗資源和性能的，因此不采用這種方式嫌套，而是使用websocket逆屡。

ajax輪詢和websocket對(duì)比.png

代碼實(shí)現(xiàn)

最上面的架構(gòu)圖就是實(shí)現(xiàn)的一個(gè)思路未蝌，我們需要將行情基礎(chǔ)數(shù)據(jù)保存到redis的一個(gè)隊(duì)列里驮吱，然后行情服務(wù)去主動(dòng)獲取隊(duì)列內(nèi)的數(shù)據(jù)，行情服務(wù)獲取到數(shù)據(jù)后萧吠，就將行情發(fā)布給網(wǎng)關(guān)左冬，然后由網(wǎng)關(guān)使用websocket將數(shù)據(jù)推送給前端用戶，同時(shí)纸型，需要將行情數(shù)據(jù)處理成k線數(shù)據(jù)拇砰，并保存到redis里以及對(duì)行情數(shù)據(jù)進(jìn)行持久化。
那么我們只要圍繞著這個(gè)思路狰腌，去實(shí)現(xiàn)就可以了：
1.將行情保存到redis隊(duì)列內(nèi)
2.行情服務(wù)獲取隊(duì)列數(shù)據(jù)
3.行情服務(wù)發(fā)布行情消息
4.行情服務(wù)處理行情數(shù)據(jù)除破，生成K線數(shù)據(jù)，并保存到redis
5.行情服務(wù)隊(duì)行情數(shù)據(jù)進(jìn)行持久化
6.網(wǎng)關(guān)訂閱行情琼腔，并使用websocet將行情推送給前端

1.將行情保存到redis隊(duì)列內(nèi)（略）

2.行情服務(wù)獲取隊(duì)列數(shù)據(jù)（分布式鎖保證順序）瑰枫，發(fā)布行情消息，生成K線數(shù)據(jù)并保存到redis丹莲，行情數(shù)據(jù)持久化

/**
 * 系統(tǒng)初始化類
 * 
 * @author Longer
 *
 */
@Component
public class SystemInitBean implements InitializingBean {

    static Logger logger = LoggerFactory.getLogger(SystemInitBean.class);
    @Autowired
    private IKLineService kLineService;
    @Autowired
    private TradeTimeService tradeTimeService;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private QuotePeriodUtil quotePeriodUtil;
    @Autowired
    private RocketMQUtil rocketMQUtil;
    @Value("${rocketmq.quotepersistenece.topics}")
    private String persistenceTopics;
    @Value("${product.list}")
    private String productList;//商品列表

    public void afterPropertiesSet() throws Exception {
        logger.info("-------------開始啟動(dòng)初始服務(wù)--------------");
        String[] productArr = productList.split(",");
        for (String productTradeNo : productArr) {
            new QuoteListListenerThread(kLineService,tradeTimeService,redisUtil,quotePeriodUtil
                    ,rocketMQUtil,persistenceTopics,productTradeNo).start();
        }
        logger.info("---------啟動(dòng)初始服務(wù)完畢-------------------");
    }
}


/**
 * @Classname QuoteListListenerThread
 * @Description 行情隊(duì)列監(jiān)聽線程
 * @Date 2019/12/23 15:43
 * @Created by Longer
 */
@Slf4j
public class QuoteListListenerThread extends Thread{
    public QuoteListListenerThread(){}
    public QuoteListListenerThread(IKLineService kLineService, TradeTimeService tradeTimeService,
                                   RedisUtil redisUtil, QuotePeriodUtil quotePeriodUtil,
                                   RocketMQUtil rocketMQUtil, String persistenceTopics, String productTradeNo){
        this.kLineService=kLineService;
        this.tradeTimeService=tradeTimeService;
        this.redisUtil=redisUtil;
        this.quotePeriodUtil=quotePeriodUtil;
        this.rocketMQUtil=rocketMQUtil;
        this.persistenceTopics=persistenceTopics;
        this.productTradeNo=productTradeNo;
    }

    private IKLineService kLineService;
    private TradeTimeService tradeTimeService;
    private RedisUtil redisUtil;
    private QuotePeriodUtil quotePeriodUtil;
    private RocketMQUtil rocketMQUtil;
    private String persistenceTopics;
    private String productTradeNo;

 @Override
    public void run() {
        while(true){
            String lockValue = UUID.randomUUID().toString();
            try{
                //加鎖（分布式鎖）光坝，不加鎖的話，就無法保證順序消費(fèi)甥材，因?yàn)橛卸鄠€(gè)節(jié)點(diǎn)
                boolean lock = redisUtil.getLock(MessageFormat.format(RedisConstant.QUOTESERVICE_KLINE_LOCK_KEY,productTradeNo)
                        , lockValue, 1);
                if(lock){
                    Object realMsg = redisUtil.rightPop(MessageFormat.format(RedisConstant.QUOTESERVICE_QUOTE_LIST,productTradeNo));
                    if(!StringUtils.isEmpty(realMsg)){
                        log.info("QuoteListListenerThread監(jiān)聽到隊(duì)列消息：{}",realMsg);
                        String s = realMsg.toString();
                        QuoteDto quoteDto = JSONObject.parseObject(s, QuoteDto.class);
                        
                        //step1.發(fā)布行情消息給網(wǎng)關(guān)
                        
                        //step2.行情數(shù)據(jù)處理（K線）和存儲(chǔ)到redis
                        
                        //step3.行情數(shù)據(jù)異步持久化
                        
                    }
                }else{
                    log.info("QuoteListListenerThread:獲取鎖失敗");
                    Thread.sleep(1 * 1000);
                }
            }catch (Exception e){
                log.info("QuoteListListenerThread報(bào)錯(cuò)：{}",e);
                try {
                    Thread.sleep(1 * 1000);
                } catch (InterruptedException e1) {
                    log.info("QuoteListListenerThread睡眠報(bào)錯(cuò)：{}",e1);
                }
            }finally {
                try {
                    //釋放鎖
                    redisUtil.releaseLock(MessageFormat.format(RedisConstant.QUOTESERVICE_KLINE_LOCK_KEY,productTradeNo),lockValue);
                }catch (Exception e){
                    log.info("釋放鎖報(bào)錯(cuò)：{}",e);
                }

            }
        }
    }
}

說明：不同的商品會(huì)對(duì)應(yīng)不同的行情盯另，所以也會(huì)將各自的行情存放到不同的隊(duì)列。這里區(qū)分商品的標(biāo)識(shí)是productTradeNo 字段（商品交易編碼）洲赵，在系統(tǒng)啟動(dòng)的時(shí)候鸳惯，會(huì)針對(duì)每一個(gè)商品，都開啟一個(gè)線程板鬓，進(jìn)行處理商品的行情悲敷。
由于行情服務(wù)是多節(jié)點(diǎn)，所以這里需要使用分布式鎖俭令，來保證每次只有一個(gè)節(jié)點(diǎn)獲取到行情數(shù)據(jù)后德，從而保證行情的順序消費(fèi)。其實(shí)如果要百分百保證順序消費(fèi)的話抄腔，最好是行情服務(wù)只部署一個(gè)節(jié)點(diǎn)瓢湃，但是這樣就不能保證行情服務(wù)的高可用了。

行情服務(wù)處理行情數(shù)據(jù)赫蛇，生成K線數(shù)據(jù)绵患，并保存到redis

剛剛在上面提過，分時(shí)線就是每一分鐘最后一口價(jià)的連線悟耘，而K線包含的數(shù)據(jù)有：“開盤價(jià)落蝙、最高價(jià)、最低價(jià)、收盤價(jià)筏勒，成交量移迫、成交額”，那么要獲取分時(shí)線數(shù)據(jù)管行，則只需要獲取1分鐘k線數(shù)據(jù)就好了厨埋，也就是以每分鐘為單位的K線數(shù)據(jù)。因?yàn)?分鐘K線的收盤價(jià)就是每分鐘的最后一口價(jià)捐顷。
那現(xiàn)在的問題就是怎么樣將基本行情數(shù)據(jù)（成交價(jià)格荡陷，成交量，成交時(shí)間）迅涮，加工處理成K線數(shù)據(jù)废赞，并且保存到redis內(nèi)。
這里就要使用到一個(gè)Java類CronSequenceGenerator
二話不說逗柴，直接上代碼蛹头。

/**
 * @Classname PeriodType
 * @Description TODO
 * @Date 2019/5/30 16:57
 * @Created by Longer
 */
public enum PeriodType {
    /** 分時(shí) */
    ONE_MINUTE("1"),

    /** 5分鐘 */
    FIVE_MINUTE("5"),

    /** 15分鐘 */
    FIFTEEN_MINUTE("15"),

    /** 30分鐘 */
    THIRTY_MINUTE("30"),

    /** 60分鐘 */
    SIXTY_MINUTE("60"),

    /** 天 */
    DAY("day"),

    /** 4小時(shí)*/
    FOUR_HOUR("4"),

    /** 周 */
    WEEK("week"),

    /** 月 */
    MONTH("month");

    private String index;

    // 構(gòu)造方法
    private PeriodType(String index) {
        this.index = index;
    }

    public String getIndex() {
        return index;
    }
}


/**
 * @Classname QuotePeriodUtil
 * @Description TODO
 * @Date 2019/5/31 14:46
 * @Created by Longer
 */
@Component
@Slf4j
public class QuotePeriodUtil {

    private static CronSequenceGenerator oneMinuteTrigger = new CronSequenceGenerator("0 0/1 * * * ? ");
    private static CronSequenceGenerator fiveMinuteTrigger = new CronSequenceGenerator("0 0/5 *  * * ? ");
    private static CronSequenceGenerator fifteenMinuteTrigger = new CronSequenceGenerator("0 0/15 *  * * ? ");
    private static CronSequenceGenerator thirtyMinuteTrigger = new CronSequenceGenerator("0 0/30 *  * * ? ");
    private static CronSequenceGenerator fourHourTrigger = new CronSequenceGenerator("0 0 0/4  * * ? ");
    private static CronSequenceGenerator sixtyMinuteTrigger = new CronSequenceGenerator("0 0/60 *  * * ? ");
    // 每天上午0：00觸發(fā)
    private static CronSequenceGenerator dayTrigger = new CronSequenceGenerator("0 0 0 * * ?");
    // 每個(gè)星期一上午0點(diǎn)觸發(fā)
    private static CronSequenceGenerator weekTrigger = new CronSequenceGenerator("0 0 0 ? * MON");
    // 表示在每月的1日的上午0點(diǎn)觸發(fā)
    private static CronSequenceGenerator monthTrigger = new CronSequenceGenerator("0 0 0 1 * ?");
    /*
    @Autowired
    private ParameterCache parameterCache;*/

    public QuotePeriod getPeriod( PeriodType type, Date dateTime) {
        QuotePeriod period = null;
        Date nextExecutionTime = null;
        switch (type) {
        case ONE_MINUTE:
            nextExecutionTime = oneMinuteTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 1 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case FIVE_MINUTE:
            nextExecutionTime = fiveMinuteTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 5 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case FIFTEEN_MINUTE:
            nextExecutionTime = fifteenMinuteTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 15 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case THIRTY_MINUTE:
            nextExecutionTime = thirtyMinuteTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 30 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case SIXTY_MINUTE:
            nextExecutionTime = sixtyMinuteTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 60 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case FOUR_HOUR:
            nextExecutionTime = fourHourTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 4 * 60 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case DAY:
            nextExecutionTime = dayTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 24 * 60 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case WEEK:
            nextExecutionTime = weekTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 7 * 24 * 60 * 60 * 1000, nextExecutionTime.getTime(), type);
            break;
        case MONTH:
            nextExecutionTime = monthTrigger.next(dateTime);
            // 獲取當(dāng)月1號(hào)上午6點(diǎn)
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(new Date(nextExecutionTime.getTime()));
            calendar.add(Calendar.MONTH, -1);// 取上個(gè)月
            calendar.set(Calendar.DAY_OF_MONTH, 1);
            calendar.set(Calendar.HOUR_OF_DAY, 0);
            calendar.set(Calendar.MINUTE, 0);
            calendar.set(Calendar.SECOND, 0);
            period = new QuotePeriod(calendar.getTimeInMillis(), nextExecutionTime.getTime(), type);
            break;
        default:
            nextExecutionTime = thirtyMinuteTrigger.next(dateTime);
            period = new QuotePeriod(nextExecutionTime.getTime() - 30 * 60 * 1000, nextExecutionTime.getTime(), PeriodType.THIRTY_MINUTE);
        }

        return period;
    }

    /**
     * 獲取當(dāng)前天的周期前的第 n 個(gè)天周期的日期時(shí)間
     * @param type
     * @param n
     * @return
     */
    public QuotePeriod getBeforePeriod( PeriodType type, int n) {
        Date nextExecutionTime = dayTrigger.next(new Date());
        QuotePeriod period = new QuotePeriod(nextExecutionTime.getTime() - 24 * 60 * 60 * 1000 * n, nextExecutionTime.getTime(), type);
        return period;
    }

使用上面我寫好的類，就可以知道當(dāng)前時(shí)間是屬于哪一分鐘戏溺，或者是屬于哪個(gè)周渣蜗，哪個(gè)月的。大家可以直接拷貝上面兩個(gè)類旷祸，然后自行進(jìn)行調(diào)試耕拷。

public static void main(String[] args) {
        QuotePeriodUtil quotePeriodUtil = new QuotePeriodUtil();
        QuotePeriod period = quotePeriodUtil.getPeriod(PeriodType.WEEK, new Date());
        System.out.println("這周的開始日期"+period.getStartTime()+"這周的結(jié)束日期"+period.getEndTime());
    }

3.網(wǎng)關(guān)訂閱行情，并使用websocet將行情推送給前端（略）

因?yàn)楸疚牡闹攸c(diǎn)是整個(gè)行情推送的設(shè)計(jì)實(shí)錄托享，并且對(duì)一些難點(diǎn)進(jìn)行說明骚烧，而ws推送的代碼網(wǎng)上是很多的，大家可以自行上網(wǎng)查找闰围，這里就不貼出來了赃绊。

總結(jié)

java實(shí)現(xiàn)股市行情實(shí)時(shí)推送,最上面的架構(gòu)圖就是實(shí)現(xiàn)的一個(gè)思路，我們需要將行情基礎(chǔ)數(shù)據(jù)保存到redis的一個(gè)隊(duì)列里羡榴，然后行情服務(wù)去主動(dòng)獲取隊(duì)列內(nèi)的數(shù)據(jù)碧查，行情服務(wù)獲取到數(shù)據(jù)后，就將行情發(fā)布給網(wǎng)關(guān)校仑，然后由網(wǎng)關(guān)使用websocket將數(shù)據(jù)推送給前端用戶忠售，同時(shí)，需要將行情數(shù)據(jù)處理成k線數(shù)據(jù)迄沫，并保存到redis里以及對(duì)行情數(shù)據(jù)進(jìn)行持久化稻扬。
那么我們只要圍繞著這個(gè)思路，去實(shí)現(xiàn)就可以了：
1.將行情保存到redis隊(duì)列內(nèi)
2.行情服務(wù)獲取隊(duì)列數(shù)據(jù)
3.行情服務(wù)發(fā)布行情消息
4.行情服務(wù)處理行情數(shù)據(jù)羊瘩，生成K線數(shù)據(jù)泰佳，并保存到redis
5.數(shù)據(jù)持久化