代码迁移推送

zj

2024-06-03 3603ecb207f7e712c635f19531e05fac4d19e53f

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
103
104
105
106
107
108
109
110
111
112
package project.data.job;
 
import java.util.Date;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.task.TaskExecutor;
 
import kernel.util.ThreadUtils;
import project.data.DataCache;
import project.data.internal.DepthTimeObject;
import project.data.internal.TradeTimeObject;
import project.data.model.Depth;
import project.data.model.Trade;
import project.hobi.HobiDataService;
import project.item.model.Item;
 
public class DataServer implements Runnable {
    
    private Logger logger = LoggerFactory.getLogger(DataServer.class);
    
    private HobiDataService hobiDataService;
 
    private TaskExecutor taskExecutor;
 
    public void start() {
        new Thread(this, "DataServer").start();
        if (logger.isInfoEnabled())
            logger.info("启动DataServer！");
    }
 
    public void run() {
        
        while (true) {
            try {
 
                HandleObject handleObject = DataQueue.poll();
 
                if (handleObject != null) {
                    this.taskExecutor.execute(new HandleRunner(handleObject, handleObject.getType()));
                } else {
                    ThreadUtils.sleep(50);
                }
 
            } catch (Throwable e) {
                logger.error("DataServer taskExecutor.execute() fail", e);
            }
        }
    }
 
    public class HandleRunner implements Runnable {
        private String type;
        private HandleObject handle;
 
        public HandleRunner(HandleObject handle, String type) {
            this.handle = handle;
            this.type = type;
        }
 
        public void run() {
            try {
                if (HandleObject.type_depth.equals(type)) {
                    depth(handle);
                } else if (HandleObject.type_trade.equals(type)) {
                    trade(handle);
                }
 
            } catch (Throwable t) {
                logger.error("HandleRunner run fail ", t);
            }
        }
 
        private void depth(HandleObject handle) {
            Item item = handle.getItem();
            Depth depth = hobiDataService.depthDecorator(item.getSymbol_data(), 0);
            if (depth != null) {
                DepthTimeObject timeObject = new DepthTimeObject();
                timeObject.setLastTime(new Date());
 
                timeObject.setDepth(depth);
                DataCache.getDepth().put(item.getSymbol(), timeObject);
            }
        }
 
        private void trade(HandleObject handle) {
            Item item = handle.getItem();
            Trade trade = hobiDataService.tradeDecorator(item.getSymbol_data(), 0);
 
            if (trade != null) {
                TradeTimeObject timeObject = DataCache.getTrade().get(item.getSymbol());
                if (timeObject == null) {
                    timeObject = new TradeTimeObject();
                }
                timeObject.setLastTime(new Date());
 
                timeObject.put(item.getSymbol(), trade.getData());
 
                DataCache.getTrade().put(item.getSymbol(), timeObject);
            }
        }
 
    }
 
    public void setTaskExecutor(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }
 
    public void setHobiDataService(HobiDataService hobiDataService) {
        this.hobiDataService = hobiDataService;
    }
 
}