feat(station): 添加策略监控中心功能

Administrator

2026-08-13 a23f23570935850192099133509e90c33769d26c

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
package com.xcong.excoin.modules.station.consumer;
 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xcong.excoin.configurations.RabbitMqConfig;
import com.xcong.excoin.modules.station.model.CmdAckMsg;
import com.xcong.excoin.modules.station.model.GateStatsEvent;
import com.xcong.excoin.modules.station.model.HeartbeatMsg;
import com.xcong.excoin.modules.station.registry.InstanceRegistry;
import com.xcong.excoin.modules.station.service.StationService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import javax.annotation.Resource;
 
/**
 * Station 消费心跳 + ACK + 策略事件 — 均复用到 QUEUE_GATE_HEARTBEAT 和 QUEUE_GATE_STATS
 */
@Slf4j
@Component
public class GateMessageConsumer {
 
    @Resource
    private InstanceRegistry instanceRegistry;
 
    @Resource
    private StationService stationService;
 
    /**
     * 消费心跳 + ACK（都路由到 heartbeat 队列）
     */
    @RabbitListener(queues = RabbitMqConfig.QUEUE_GATE_HEARTBEAT)
    public void onHeartbeat(String msg) {
        try {
            JSONObject json = JSON.parseObject(msg);
            String type = json.getString("type");
            String apiKeyMd5 = json.getString("apiKeyMd5");
 
            if ("HEARTBEAT".equals(type)) {
                HeartbeatMsg hb = JSON.parseObject(json.getString("payload"), HeartbeatMsg.class);
                instanceRegistry.update(apiKeyMd5, hb);
                log.debug("[Station] 心跳: {}, state={}", apiKeyMd5, hb.getState());
 
            } else if ("CMD_ACK".equals(type)) {
                CmdAckMsg ack = JSON.parseObject(json.getString("payload"), CmdAckMsg.class);
                instanceRegistry.updateState(apiKeyMd5, ack.getNewState());
                log.info("[Station] 指令确认: cmdId={}, success={}, newState={}",
                        ack.getCommandId(), ack.isSuccess(), ack.getNewState());
            }
 
        } catch (Exception e) {
            log.error("[Station] 心跳消费异常", e);
        }
    }
 
    /**
     * 消费策略事件（stats 队列，落库）
     */
    @RabbitListener(queues = RabbitMqConfig.QUEUE_GATE_STATS)
    public void onStats(String msg) {
        try {
            GateStatsEvent event = JSON.parseObject(msg, GateStatsEvent.class);
            stationService.saveEvent(event);
        } catch (Exception e) {
            log.error("[Station] 策略事件消费异常", e);
        }
    }
}