autumn/jsowell-charger-web
1
0
Fork 0
mirror of https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git synced 2026-04-20 19:15:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

update mqtt 服务器

Browse Source
This commit is contained in:
Lemon
2023-12-21 10:37:21 +08:00
parent 5fbce62752
commit 36c2832f63
6 changed files with 43 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServer.java
View File

@@ -64,41 +64,9 @@ public class NettyServer implements CommandLineRunner {
} else {
log.error("NettyServer启动失败", future.cause());
}
//
// ServerBootstrap mqttBootstrap = new ServerBootstrap();
// mqttBootstrap.group(bossGroup, workerGroup);
// mqttBootstrap.channel(NioServerSocketChannel.class);
//
// mqttBootstrap.option(ChannelOption.SO_REUSEADDR, true)
// .option(ChannelOption.SO_BACKLOG, 1024)
// .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
// .option(ChannelOption.SO_RCVBUF, 10485760);
//
// mqttBootstrap.childOption(ChannelOption.TCP_NODELAY, true)
// .childOption(ChannelOption.SO_KEEPALIVE, true)
// .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
//
// mqttBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
// protected void initChannel(SocketChannel ch) {
// ChannelPipeline channelPipeline = ch.pipeline();
// // 设置读写空闲超时时间
// channelPipeline.addLast(new IdleStateHandler(600, 600, 1200));
// channelPipeline.addLast("encoder", MqttEncoder.INSTANCE);
// channelPipeline.addLast("decoder", new MqttDecoder());
// channelPipeline.addLast(new BootNettyMqttChannelInboundHandler());
// }
// });
// ChannelFuture future2 = mqttBootstrap.bind(address2.getPort()).sync();
// if(future2.isSuccess()){
// log.info("MqttServer启动成功, 开始监听端口:{}", address2.getPort());
// future2.channel().closeFuture().sync();
// } else {
// log.error("MqttServer启动失败", future2.cause());
// }
//关闭channel和块，直到它被关闭
future.channel().closeFuture().sync();
// future2.channel().closeFuture().sync();
// 关闭channel和块，直到它被关闭 （若取消注释，则 MQTT 服务器无法启动）
// future.channel().closeFuture().sync();
} catch (Exception e) {
log.error("NettyServer.start error", e);
bossGroup.shutdownGracefully();

17
jsowell-netty/src/main/java/com/jsowell/netty/service/camera/CameraBusinessService.java
View File

@@ -12,9 +12,22 @@ import java.net.UnknownHostException;
* @Date 2023/12/20 15:15:26
*/
public interface CameraBusinessService {
public void sendGroundLockCommand(String sn, String msgType, String msgPrefix, String topic) throws InterruptedException;
/**
* 发送具体指令到某主题
* @param sn 设备 sn
* @param msgType 消息类型
* @param msgPrefix 消息前缀
* @param topic 主题
* @param msgData 消息内容
* @throws InterruptedException
*/
public void sendGroundLockCommand(String sn, String msgType, String msgPrefix, String topic, String msgData) throws InterruptedException;
/**
* 解析channel中的ip地址， 并将 sn 和 channelId 进行绑定，存入缓存
* @param channel
* @throws UnknownHostException
*/
public void processConnectMsg(Channel channel) throws UnknownHostException;
}

28
jsowell-netty/src/main/java/com/jsowell/netty/service/camera/impl/CameraBusinessServiceImpl.java
View File

@@ -4,14 +4,11 @@ import com.alibaba.fastjson2.JSONObject;
import com.jsowell.common.constant.CacheConstants;
import com.jsowell.common.core.redis.RedisCache;
import com.jsowell.common.util.DateUtils;
import com.jsowell.common.util.StringUtils;
import com.jsowell.common.util.sign.MD5Util;
import com.jsowell.netty.domain.MqttRequest;
import com.jsowell.netty.server.mqtt.BootNettyMqttChannelInboundHandler;
import com.jsowell.netty.service.camera.CameraBusinessService;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.handler.codec.mqtt.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@@ -34,20 +31,37 @@ public class CameraBusinessServiceImpl implements CameraBusinessService {
@Autowired
private BootNettyMqttChannelInboundHandler handler;
public void sendGroundLockCommand(String sn, String msgType, String msgPrefix, String topic) throws InterruptedException {
/**
* 发送具体指令到某主题
* @param sn 设备 sn
* @param msgType 消息类型
* @param msgPrefix 消息前缀
* @param topic 主题
* @param msgData 消息内容
* @throws InterruptedException
*/
public void sendGroundLockCommand(String sn, String msgType, String msgPrefix, String topic, String msgData) throws InterruptedException {
JSONObject jsonObject = spliceStr(sn, msgType, msgPrefix);
// 通过sn查找出对应的channelId
String mqttConnectRedisKey = CacheConstants.MQTT_CONNECT_CHANNEL + sn;
String mqttConnectRedisKey = CacheConstants.MQTT_CONNECT_SN + sn;
Object cacheObject = redisCache.getCacheObject(mqttConnectRedisKey);
if (cacheObject == null) {
return;
}
String channelId = (String) cacheObject;
if (StringUtils.isNotBlank(msgData)) {
jsonObject.put("msg_data", msgData);
}
// 发送消息
handler.sendMsg(channelId, topic, jsonObject.toJSONString());
}
/**
* 解析channel中的ip地址， 并将 sn 和 channelId 进行绑定，存入缓存
* @param channel
* @throws UnknownHostException
*/
public void processConnectMsg(Channel channel) throws UnknownHostException {
// 解析 channel 中的 ip 地址
String remoteAddress = channel.remoteAddress().toString();
@@ -68,7 +82,7 @@ public class CameraBusinessServiceImpl implements CameraBusinessService {
if (cacheObject != null) {
String sn = (String) cacheObject;
// 将 sn 和 channelId 绑定关系存入缓存
String mqttConnectRedisKey = CacheConstants.MQTT_CONNECT_CHANNEL + sn;
String mqttConnectRedisKey = CacheConstants.MQTT_CONNECT_SN + sn;
redisCache.setCacheObject(mqttConnectRedisKey, channel.id().asShortText());
}
}