change - broker集群订阅Redis实现消息下发
parent
689c39d829
commit
a36ab5537e
@ -1,25 +0,0 @@
|
||||
package com.hw.mqtt.domain;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* @author WenJY
|
||||
* @date 2023年03月14日 13:46
|
||||
*/
|
||||
@Data
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class ServerNode {
|
||||
|
||||
/**
|
||||
* 节点名称
|
||||
*/
|
||||
private String name;
|
||||
/**
|
||||
* ip:port
|
||||
*/
|
||||
private String peerHost;
|
||||
|
||||
}
|
||||
@ -1,62 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.hw.mqtt.service;
|
||||
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* mqtt broker 服务
|
||||
*
|
||||
* @author L.cm
|
||||
*/
|
||||
public interface IMqttBrokerService {
|
||||
|
||||
/**
|
||||
* 获取在线客户端数量
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 13:51
|
||||
* @return long
|
||||
*/
|
||||
long getOnlineClientSize();
|
||||
|
||||
/**
|
||||
* 获取所有在线的客户端
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 13:51
|
||||
* @return java.util.List<java.lang.String>
|
||||
*/
|
||||
List<String> getOnlineClients();
|
||||
|
||||
/**
|
||||
* 向指定主题发送消息
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 13:50
|
||||
* @param topic
|
||||
* @param payload
|
||||
* @return boolean
|
||||
*/
|
||||
boolean publish(String topic,String payload);
|
||||
|
||||
/**
|
||||
* 主动关闭指定客户端连接
|
||||
* @author WenJY
|
||||
* @date 2023-03-18 9:23
|
||||
* @param clientId
|
||||
*/
|
||||
boolean closeClientById(String clientId);
|
||||
}
|
||||
@ -1,96 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.hw.mqtt.service.impl;
|
||||
|
||||
import com.hw.mqtt.enums.RedisKeys;
|
||||
import com.hw.mqtt.service.IMqttBrokerService;
|
||||
import net.dreamlu.iot.mqtt.spring.server.MqttServerTemplate;
|
||||
import net.dreamlu.mica.core.utils.StringPool;
|
||||
import net.dreamlu.mica.redis.cache.MicaRedisCache;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* mqtt broker 服务
|
||||
*
|
||||
* @author L.cm
|
||||
*/
|
||||
@Service
|
||||
public class MqttBrokerServiceImpl implements IMqttBrokerService {
|
||||
private static final Logger logger = LoggerFactory.getLogger(MqttBrokerServiceImpl.class);
|
||||
@Autowired private MicaRedisCache redisCache;
|
||||
@Autowired private MqttServerTemplate server;
|
||||
|
||||
@Override
|
||||
public long getOnlineClientSize() {
|
||||
Set<String> keySet = redisCache.scan(RedisKeys.CONNECT_STATUS.getKey(StringPool.STAR));
|
||||
if (keySet.isEmpty()) {
|
||||
return 0L;
|
||||
}
|
||||
long result = 0;
|
||||
for (String redisKey : keySet) {
|
||||
Long count = redisCache.getSetOps().size(redisKey);
|
||||
if (count != null) {
|
||||
result += count;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> getOnlineClients() {
|
||||
Set<String> keySet = redisCache.scan(RedisKeys.CONNECT_STATUS.getKey(StringPool.STAR));
|
||||
if (keySet.isEmpty()) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<String> clientList = new ArrayList<>();
|
||||
for (String redisKey : keySet) {
|
||||
Set<String> members = redisCache.sMembers(redisKey);
|
||||
if (members != null && !members.isEmpty()) {
|
||||
clientList.addAll(members);
|
||||
}
|
||||
}
|
||||
return clientList;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean publish(String topic, String payload) {
|
||||
boolean result = server.publishAll(topic, payload.getBytes(StandardCharsets.UTF_8));
|
||||
logger.info("Mqtt publishAll result:{};topic:{};payload:{}", result, topic, payload);
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean closeClientById(String clientId) {
|
||||
try{
|
||||
server.close(clientId);
|
||||
}catch (Exception ex){
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -1,40 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.hw.mqtt.util;
|
||||
|
||||
import net.dreamlu.mica.core.utils.CharPool;
|
||||
|
||||
/**
|
||||
* redis 工具
|
||||
*
|
||||
* @author L.cm
|
||||
*/
|
||||
public class RedisUtil {
|
||||
|
||||
/**
|
||||
* 转换成 redis 的 pattern 规则
|
||||
*
|
||||
* @return pattern
|
||||
*/
|
||||
public static String getTopicPattern(String topicFilter) {
|
||||
// mqtt 分享主题 $share/{ShareName}/{filter}
|
||||
return topicFilter
|
||||
.replace(CharPool.PLUS, CharPool.STAR)
|
||||
.replace(CharPool.HASH, CharPool.STAR);
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue