业主相关 功能 从listener 迁移为 cmd 模式

java110

2022-06-21 e62a227ec48c9ee1f738a469480c6e59d436ed3d

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
package com.java110.api.kafka;
 
import com.alibaba.fastjson.JSONObject;
import com.java110.api.websocket.MessageWebsocket;
import com.java110.api.websocket.ParkingAreaWebsocket;
import com.java110.core.base.controller.BaseController;
import com.java110.utils.constant.KafkaConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import com.java110.core.log.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
 
/**
 * kafka侦听
 * Created by wuxw on 2018/4/15.
 */
public class FrontServiceKafka extends BaseController {
 
    private final static Logger logger = LoggerFactory.getLogger(FrontServiceKafka.class);
 
 
    /**
     * 像前段返回内容
     *
     * @param record
     */
    @KafkaListener(topics = {KafkaConstant.TOPIC_API_SEND_WEB})
    public void listen(ConsumerRecord<?, ?> record) {
        logger.info("kafka的key: " + record.key());
        logger.info("kafka的value: " + record.value().toString());
 
        JSONObject param = null;
        try {
            param = JSONObject.parseObject(record.value().toString());
            MessageWebsocket.sendInfo(param.toJSONString(), param.getString("userId"));
        } catch (Exception e) {
            logger.error("发送消息失败", e);
        } finally {
 
        }
    }
 
    /**
     * 像前段返回内容
     *
     * @param record
     */
    @KafkaListener(topics = {KafkaConstant.TOPIC_API_SEND_PARKING_AREA_WEB})
    public void listenParkingArea(ConsumerRecord<?, ?> record) {
        logger.info("kafka的key: " + record.key());
        logger.info("kafka的value: " + record.value().toString());
 
        JSONObject param = null;
        try {
            param = JSONObject.parseObject(record.value().toString());
            ParkingAreaWebsocket.sendInfo(param.toJSONString(), param.getString("extBoxId"));
        } catch (Exception e) {
            logger.error("发送消息失败", e);
        } finally {
 
        }
    }
 
 
}