BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 4be18af236ac036741c6adba23fa58d06f17a73c

Authored by puhui999
1 parent 04009322

【代码优化】IoT: 基于 guava 对 producer 做 cache

Inline Side-by-side
Showing 1 changed file with 50 additions and 19 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/execute/IotRocketMQDataBridgeExecute.java
  View file @4be18af
... ... @@ -3,6 +3,9 @@ package cn.iocoder.yudao.module.iot.service.rule.execute;
3 3 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataBridgeDO;
4 4 import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgTypeEnum;
5 5 import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
  6 +import com.google.common.cache.CacheBuilder;
  7 +import com.google.common.cache.CacheLoader;
  8 +import com.google.common.cache.LoadingCache;
6 9 import lombok.extern.slf4j.Slf4j;
7 10 import org.apache.rocketmq.client.producer.DefaultMQProducer;
8 11 import org.apache.rocketmq.client.producer.SendResult;
... ... @@ -11,7 +14,9 @@ import org.apache.rocketmq.common.message.Message;
11 14 import org.apache.rocketmq.remoting.common.RemotingHelper;
12 15 import org.springframework.stereotype.Component;
13 16  
  17 +import java.time.Duration;
14 18 import java.time.LocalDateTime;
  19 +import java.util.concurrent.Executors;
15 20  
16 21 /**
17 22 * RocketMQ 的 {@link IotDataBridgeExecute} 实现类
... ... @@ -22,6 +27,36 @@ import java.time.LocalDateTime;
22 27 @Slf4j
23 28 public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
24 29  
  30 + /**
  31 + * 针对 {@link IotDataBridgeDO.RocketMQConfig} 的 DefaultMQProducer 缓存
  32 + */
  33 + private final LoadingCache<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer> PRODUCER_CACHE = CacheBuilder.newBuilder()
  34 + // 只阻塞当前数据加载线程，其他线程返回旧值
  35 + .refreshAfterWrite(Duration.ofMinutes(10))
  36 + // 增加移除监听器，自动关闭 producer
  37 + .removalListener(notification -> {
  38 + DefaultMQProducer producer = (DefaultMQProducer) notification.getValue();
  39 + if (producer != null) {
  40 + try {
  41 + producer.shutdown();
  42 + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey());
  43 + } catch (Exception e) {
  44 + log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e);
  45 + }
  46 + }
  47 + })
  48 + // 通过 asyncReloading 实现全异步加载，包括 refreshAfterWrite 被阻塞的加载线程
  49 + .build(CacheLoader.asyncReloading(new CacheLoader<IotDataBridgeDO.RocketMQConfig, DefaultMQProducer>() {
  50 + @Override
  51 + public DefaultMQProducer load(IotDataBridgeDO.RocketMQConfig config) throws Exception {
  52 + DefaultMQProducer producer = new DefaultMQProducer(config.getGroup());
  53 + producer.setNamesrvAddr(config.getNameServer());
  54 + producer.start();
  55 + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config);
  56 + return producer;
  57 + }
  58 + }, Executors.newCachedThreadPool()));
  59 +
25 60 @Override
26 61 public void execute(IotDeviceMessage message, IotDataBridgeDO dataBridge) {
27 62 // 1.1 校验数据桥接的类型 == ROCKETMQ
... ... @@ -33,14 +68,9 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
33 68 }
34 69  
35 70 private void executeRocketMQ(IotDeviceMessage message, IotDataBridgeDO.RocketMQConfig config) {
36   - // 1.1 创建生产者实例，指定生产者组名
37   - DefaultMQProducer producer = new DefaultMQProducer(config.getGroup());
38   - // TODO @puhui999：可以考虑，基于 guava 做 cache，使用 config 作为 key，然后假个 listener 超时，销毁 producer
39 71 try {
40   - // 1.2 设置 NameServer 地址
41   - producer.setNamesrvAddr(config.getNameServer());
42   - // 1.3 启动生产者
43   - producer.start();
  72 + // 1. 获取或创建 Producer
  73 + DefaultMQProducer producer = PRODUCER_CACHE.get(config);
44 74  
45 75 // 2.1 创建消息对象，指定Topic、Tag和消息体
46 76 Message msg = new Message(
... ... @@ -58,18 +88,22 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
58 88 }
59 89 } catch (Exception e) {
60 90 log.error("[executeRocketMQ][message({}) config({}) 发送异常]", message, config, e);
61   - } finally {
62   - // 3. 关闭生产者
63   - producer.shutdown();
64 91 }
65 92 }
66 93  
67 94 // TODO @芋艿：测试代码，后续清理
68 95 public static void main(String[] args) {
69   - // 1. 创建 IotRocketMQDataBridgeExecute 实例
  96 + // 1. 创建一个共享的实例
70 97 IotRocketMQDataBridgeExecute action = new IotRocketMQDataBridgeExecute();
71 98  
72   - // 2. 创建测试消息
  99 + // 2. 创建共享的配置
  100 + IotDataBridgeDO.RocketMQConfig config = new IotDataBridgeDO.RocketMQConfig();
  101 + config.setNameServer("127.0.0.1:9876");
  102 + config.setGroup("test-group");
  103 + config.setTopic("test-topic");
  104 + config.setTags("test-tag");
  105 +
  106 + // 3. 创建共享的消息
73 107 IotDeviceMessage message = IotDeviceMessage.builder()
74 108 .requestId("TEST-001")
75 109 .productKey("testProduct")
... ... @@ -82,14 +116,11 @@ public class IotRocketMQDataBridgeExecute implements IotDataBridgeExecute {
82 116 .tenantId(1L)
83 117 .build();
84 118  
85   - // 3. 创建 RocketMQ 配置
86   - IotDataBridgeDO.RocketMQConfig config = new IotDataBridgeDO.RocketMQConfig();
87   - config.setNameServer("127.0.0.1:9876");
88   - config.setGroup("test-group");
89   - config.setTopic("test-topic");
90   - config.setTags("test-tag");
  119 + // 4. 执行两次测试，验证缓存
  120 + log.info("[main][第一次执行，应该会创建新的 producer]");
  121 + action.executeRocketMQ(message, config);
91 122  
92   - // 4. 执行测试
  123 + log.info("[main][第二次执行，应该会复用缓存的 producer]");
93 124 action.executeRocketMQ(message, config);
94 125 }
95 126  
... ...