文章目录
- 1. 引言
- 2. 延时关机的实现方式
- 2.1 自建定时服务实现
- 2.2 RocketMQ中间件实现
- 2.2.1 生成端demo
- 2.2.2 消费端demo
- 3. 结尾
1. 引言
在设备联动中，有些场景需要保持设备继续工作一段时间再关机。比如在厨房场景下，存在燃气灶和烟机的联动场景，以燃气灶为条件，烟机为结果。当打开燃气灶后，烟机自动打开；当关掉燃气灶后，烟机需要延时3分钟再关机。
物联网设备，打开或者关闭燃气灶的时候，设备的状态会进行上报，当云端收到状态变化信号后，就能去触发烟机的动作。对于设备的联动链路可以阅读之前的文章：
那对于延时关机，就不能收到状态变化后，直接就去控制设备啦。这时候就需要定时服务来介入。
2. 延时关机的实现方式
2.1 自建定时服务实现
在微服务的时候，很多的项目根据业务的不同划分了单独的服务，因此，在物联网领域就需要有个定时服务，这个服务就只做定时；当要做延时时，就向定时服务中插入一条带有未来时间的数据，当定时服务轮询到该数据时，就执行回调，再执行设备控制。具体的时序图如下：
在这张图中，共有4个模块，条件-设备作为条件设备，进行状态上报；结果-设备作为触发设备，是被控制设备；主服务，即进行逻辑判读和设备控制的服务；定时作为任务轮询，到期回调。
具体的判断逻辑：当时设备-条件上报关机到主服务，主服务进行逻辑判断是否需要延时关机，如果不需要就直接控制设备，当需要延时关机时，就插入预约任务到到定时任务，主服务等待定时服务的回调，当收到回调后，再去控制结果-设备。
2.2 RocketMQ中间件实现
RocketMQ作为分布式消息中间件，深受各个行业的广泛使用，之前就介绍过物联网中RocketMQ的使用。在之前的介绍中，主要集中于消费端那一块，但其生产端的功能也很多，比如可以发送延时消息。因此当设备需要延时关机时，就可以利用到延时消息功能，在生产端输入控制参数的时候，加上一个时间戳，当这个时间戳到了之后，消费者才进行设备控制，这样就实现了设备的延时控制。
2.2.1 生成端demo
为了发送一个延时消息，RocketMQ提供了delayTimeLevel属性，对于生产端的实现方式：
其工作流程如下：
其代码demo如下：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class DelayedMessageProducer {
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("example_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();
// Create a message with a 10 second delay
Message message = new Message("example_topic", "Hello, RocketMQ!".getBytes(RemotingHelper.DEFAULT_CHARSET));
message.setDelayTimeLevel(3);
// Send the message
producer.send(message);
producer.shutdown();
}
}
在RocketMQ中提供了多个延时级别：
1：1s
2：5s
3：10s
4：30s
5：1m
6：2m
7：3m
8：4m
9：5m
10：6m
11：7m
12：8m
13：9m
14：10m
15：20m
16：30m
17：1h
如果要发送一个延迟 3 分钟的消息，可以设置 DelayTimeLevel 属性为 7，即message.setDelayTimeLevel(7)。
2.2.2 消费端demo
在消费者端，就需要去监听，延时消息是否已经到了，如果到了就进行消费，如果没有到就进行等待。
其工作流程如下：
其代码demo如下：
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class DelayedMessageConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("example_topic", "*");
// Set the message listener
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
for (MessageExt message : messages) {
// Check if message is delayed
if (message.getProperty("DELAY_TIME_LEVEL") != null) {
long delayTimeLevel = Long.parseLong(message.getProperty("DELAY_TIME_LEVEL"));
long diff = System.currentTimeMillis() - message.getStoreTimestamp();
if (diff >= delayTimeLevel * 1000) {
// Process the message
System.out.println("Received delayed message: " + new String(message.getBody()));
// todo 实现延时控制设备
} else {
// Re-consume the message later
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
} else {
// Process the message
System.out.println("Received message: " + new String(message.getBody()));
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer started");
}
}
在上面的代码中，创建一个DefaultMQPushConsumer的实例，并且订阅监听topic；在registerMessageListener方法中进行逻辑判断，延时时间是否已经到了，如果到了就进行消费，如果没有到就进行等待。
3. 结尾
综上所述，介绍了两种实现延时关机的方式，第一种需要自己建立一个单独的定时服务，需要自己开发，难度较大，但是灵活性很高，可以自定义规则；第二种是利用了RocketMQ中延时消息方式实现的，其延时时间是固定的，不支持自定义。对于量很大的设备控制，其上的两种方式都需要搭建集群方式，保障系统的稳定性。
