mirror of
https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git
synced 2026-04-21 03:25:12 +08:00
新增 netty整合mqtt协议，与车位相机通讯并保存通讯信息
This commit is contained in:
Lemon
@@ -1,10 +1,274 @@
|
||||
package com.jsowell.netty.server.mqtt;
|
||||
|
||||
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import com.jsowell.common.util.StringUtils;
|
||||
import com.jsowell.common.util.bean.BeanUtils;
|
||||
import com.jsowell.netty.domain.MqttRequest;
|
||||
import com.jsowell.netty.service.camera.CameraBusinessService;
|
||||
import com.jsowell.netty.service.camera.impl.CameraBusinessServiceImpl;
|
||||
import com.jsowell.thirdparty.camera.service.CameraService;
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.handler.codec.mqtt.*;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* TODO
|
||||
* MQTT服务端I/O数据读写处理类
|
||||
*
|
||||
* @author Lemon
|
||||
* @Date 2023/12/14 11:33:03
|
||||
*/
|
||||
public class BootNettyMqttChannelInboundHandler {
|
||||
|
||||
@ChannelHandler.Sharable
|
||||
@Slf4j
|
||||
@Component
|
||||
public class BootNettyMqttChannelInboundHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
/**
|
||||
* Key: channelId
|
||||
* Value: ctx
|
||||
*/
|
||||
private static final ConcurrentHashMap<String, ChannelHandlerContext> CHANNEL_MAP = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* Key: channelId
|
||||
* Value: 客户端id
|
||||
*/
|
||||
private static final ConcurrentHashMap<String, String> CLIENT_MAP = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* 从客户端收到新的数据时，这个方法会在收到消息时被调用
|
||||
*/
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
|
||||
if (null != msg) {
|
||||
MqttMessage mqttMessage = (MqttMessage) msg;
|
||||
log.info("MqttServer收到消息:{}", mqttMessage.toString());
|
||||
MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
|
||||
Channel channel = ctx.channel();
|
||||
|
||||
if (mqttFixedHeader.messageType().equals(MqttMessageType.CONNECT)) {
|
||||
// 在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
|
||||
// to do 建议connect消息单独处理，用来对客户端进行认证管理等 这里先直接返回一个CONNACK消息
|
||||
BootNettyMqttMsgBack.connack(channel, mqttMessage);
|
||||
// 如果map中不包含此连接，就保存连接
|
||||
String channelId = channel.id().asShortText();
|
||||
System.out.println(channelId);
|
||||
if (CHANNEL_MAP.get(channelId) != null) {
|
||||
log.info("客户端【{}】是连接状态，连接通道数量: {}", channelId, CHANNEL_MAP.size());
|
||||
} else {
|
||||
//保存连接
|
||||
CHANNEL_MAP.put(channelId, ctx);
|
||||
// System.out.println(channel.remoteAddress()); // /192.168.2.100:39396
|
||||
log.info("客户端【{}】, 连接Mqtt服务器, 连接通道数量: {}", channelId, CHANNEL_MAP.size());
|
||||
CameraBusinessService cameraBusinessService = BeanUtils.getBean(CameraBusinessService.class);
|
||||
cameraBusinessService.processConnectMsg(channel);
|
||||
}
|
||||
// 保存客户端id
|
||||
String clientIdentifier = parsePayLoadInfo(mqttMessage.payload().toString());
|
||||
if (StringUtils.isNotBlank(clientIdentifier)) {
|
||||
CLIENT_MAP.put(channelId, clientIdentifier);
|
||||
}
|
||||
}
|
||||
|
||||
switch (mqttFixedHeader.messageType()) {
|
||||
case PUBLISH: // 客户端发布消息
|
||||
// PUBACK报文是对QoS 1等级的PUBLISH报文的响应
|
||||
System.out.println("客户端发布消息");
|
||||
BootNettyMqttMsgBack.puback(channel, mqttMessage);
|
||||
break;
|
||||
case PUBREL: // 发布释放
|
||||
// PUBREL报文是对PUBREC报文的响应
|
||||
// to do
|
||||
BootNettyMqttMsgBack.pubcomp(channel, mqttMessage);
|
||||
break;
|
||||
case SUBSCRIBE: // 客户端订阅主题
|
||||
// 客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅，每个订阅注册客户端关心的一个或多个主题。
|
||||
// 为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。
|
||||
// SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端
|
||||
// to do
|
||||
BootNettyMqttMsgBack.suback(channel, mqttMessage);
|
||||
break;
|
||||
case UNSUBSCRIBE: // 客户端取消订阅
|
||||
// 客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题
|
||||
// to do
|
||||
BootNettyMqttMsgBack.unsuback(channel, mqttMessage);
|
||||
break;
|
||||
case PINGREQ: // 客户端发起心跳
|
||||
// 客户端发送PINGREQ报文给服务端的
|
||||
// 在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着
|
||||
// 请求服务端发送 响应确认它还活着，使用网络以确认网络连接没有断开
|
||||
BootNettyMqttMsgBack.pingresp(channel, mqttMessage);
|
||||
break;
|
||||
case DISCONNECT: // 客户端主动断开连接
|
||||
// DISCONNECT报文是客户端发给服务端的最后一个控制报文， 服务端必须验证所有的保留位都被设置为0
|
||||
// to do
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从客户端收到新的数据、读取完成时调用
|
||||
*/
|
||||
@Override
|
||||
public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端与服务端第一次建立连接时执行 在channelActive方法之前执行
|
||||
*/
|
||||
@Override
|
||||
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
|
||||
super.channelRegistered(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端与服务端 断连时执行 channelInactive方法之后执行
|
||||
*/
|
||||
@Override
|
||||
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
|
||||
super.channelUnregistered(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
|
||||
*/
|
||||
@Override
|
||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
||||
super.exceptionCaught(ctx, cause);
|
||||
ctx.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端与服务端第一次建立连接时执行
|
||||
*/
|
||||
@Override
|
||||
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
||||
super.channelActive(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端与服务端 断连时执行
|
||||
*/
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
|
||||
// 删除连接
|
||||
removeConnect(ctx);
|
||||
super.channelInactive(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* 服务端 当读超时时 会调用这个方法
|
||||
*/
|
||||
@Override
|
||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
|
||||
// 删除连接
|
||||
removeConnect(ctx);
|
||||
super.userEventTriggered(ctx, evt);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
|
||||
super.channelWritabilityChanged(ctx);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* 删除连接
|
||||
* @param ctx
|
||||
*/
|
||||
public void removeConnect(ChannelHandlerContext ctx) {
|
||||
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||
String clientIp = insocket.getAddress().getHostAddress();
|
||||
String channelId = ctx.channel().id().asShortText();
|
||||
//包含此客户端才去删除
|
||||
if (CHANNEL_MAP.get(channelId) != null) {
|
||||
//删除连接
|
||||
CHANNEL_MAP.remove(channelId);
|
||||
CLIENT_MAP.remove(channelId);
|
||||
log.info("客户端【{}】, 退出Mqtt服务器[IP:{}--->PORT:{}], 连接通道数量: {}", channelId, clientIp, insocket.getPort(), CHANNEL_MAP.size());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析 payload 字段，并将 clientIdentifier （客户端id）返回
|
||||
* @param payloadText
|
||||
* @return 客户端id
|
||||
*/
|
||||
private String parsePayLoadInfo(String payloadText) {
|
||||
if (StringUtils.isBlank(payloadText)) {
|
||||
return null;
|
||||
}
|
||||
JSONObject payload = new JSONObject();
|
||||
// 使用逗号和等号进行分割
|
||||
String[] parts = payloadText.split("[,\\[\\]]");
|
||||
for (String part : parts) {
|
||||
String[] keyValue = part.split("=");
|
||||
|
||||
if (keyValue.length == 2) {
|
||||
String key = keyValue[0].trim();
|
||||
String value = keyValue[1].trim();
|
||||
|
||||
payload.put(key, value);
|
||||
}
|
||||
}
|
||||
return payload.getString("clientIdentifier");
|
||||
}
|
||||
|
||||
/**
|
||||
* 通过 channelId 获取 Channel
|
||||
* @param channelId
|
||||
* @return
|
||||
*/
|
||||
public Channel getChannel(String channelId) {
|
||||
if (channelId != null) {
|
||||
ChannelHandlerContext ctx = CHANNEL_MAP.get(channelId);
|
||||
return ctx.channel();
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 服务器端向某Topic发送消息
|
||||
* @param channelId 发送的通道id
|
||||
* @param topic 主题Topic
|
||||
* @param requestMsg 要发送的Json对象
|
||||
* @return
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public ChannelFuture sendMsg(String channelId, String topic, String requestMsg) throws InterruptedException {
|
||||
MqttRequest request = new MqttRequest((requestMsg.toString().getBytes()));
|
||||
MqttPublishMessage pubMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
|
||||
new MqttFixedHeader(MqttMessageType.PUBLISH,
|
||||
request.isDup(),
|
||||
request.getQos(),
|
||||
request.isRetained(),
|
||||
0),
|
||||
new MqttPublishVariableHeader(topic, 0),
|
||||
Unpooled.buffer().writeBytes(request.getPayload()));
|
||||
|
||||
Channel channel = getChannel(channelId);
|
||||
// 超过高水位，则采取同步模式
|
||||
if (channel.isWritable()) {
|
||||
return channel.writeAndFlush(pubMessage);
|
||||
}
|
||||
return channel.writeAndFlush(pubMessage).sync();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,10 +1,188 @@
|
||||
package com.jsowell.netty.server.mqtt;
|
||||
|
||||
import java.net.InetAddress;
|
||||
import java.net.UnknownHostException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import com.jsowell.common.core.redis.RedisCache;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
|
||||
import io.netty.handler.codec.mqtt.MqttConnectMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
|
||||
import io.netty.handler.codec.mqtt.MqttConnectVariableHeader;
|
||||
import io.netty.handler.codec.mqtt.MqttFixedHeader;
|
||||
import io.netty.handler.codec.mqtt.MqttMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
|
||||
import io.netty.handler.codec.mqtt.MqttMessageType;
|
||||
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttPublishMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttQoS;
|
||||
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttSubAckPayload;
|
||||
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* TODO
|
||||
*
|
||||
* @author Lemon
|
||||
* @Date 2023/12/14 11:36:30
|
||||
*/
|
||||
@Component
|
||||
public class BootNettyMqttMsgBack {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(BootNettyMqttMsgBack.class);
|
||||
|
||||
/**
|
||||
* 确认连接请求
|
||||
*
|
||||
* @param channel
|
||||
* @param mqttMessage
|
||||
*/
|
||||
public static void connack(Channel channel, MqttMessage mqttMessage) {
|
||||
log.info("服务器端收到确认连接请求：{}", mqttMessage.toString());
|
||||
MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
|
||||
MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
|
||||
MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
|
||||
|
||||
// 构建返回报文， 可变报头
|
||||
MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
|
||||
// 构建返回报文， 固定报头
|
||||
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
|
||||
// 构建CONNACK消息体
|
||||
MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
|
||||
log.info("服务器端回复确认连接请求：{}", connAck.toString());
|
||||
channel.writeAndFlush(connAck);
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 根据qos发布确认
|
||||
*
|
||||
* @param channel
|
||||
* @param mqttMessage
|
||||
*/
|
||||
public static void puback(Channel channel, MqttMessage mqttMessage) {
|
||||
MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
|
||||
MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
|
||||
MqttQoS qos = (MqttQoS) mqttFixedHeaderInfo.qosLevel();
|
||||
byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
|
||||
mqttPublishMessage.payload().readBytes(headBytes);
|
||||
String data = new String(headBytes);
|
||||
System.out.println("publish data--" + data);
|
||||
|
||||
switch (qos) {
|
||||
case AT_MOST_ONCE: // 至多一次
|
||||
break;
|
||||
case AT_LEAST_ONCE: // 至少一次
|
||||
// 构建返回报文， 可变报头
|
||||
MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
|
||||
// 构建返回报文， 固定报头
|
||||
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
|
||||
// 构建PUBACK消息体
|
||||
MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
|
||||
log.info("back--" + pubAck.toString());
|
||||
channel.writeAndFlush(pubAck);
|
||||
break;
|
||||
case EXACTLY_ONCE: // 刚好一次
|
||||
// 构建返回报文， 固定报头
|
||||
MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
|
||||
// 构建返回报文， 可变报头
|
||||
MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
|
||||
MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
|
||||
log.info("back--" + mqttMessageBack.toString());
|
||||
channel.writeAndFlush(mqttMessageBack);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布完成 qos2
|
||||
*
|
||||
* @param channel
|
||||
* @param mqttMessage
|
||||
*/
|
||||
public static void pubcomp(Channel channel, MqttMessage mqttMessage) {
|
||||
MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
|
||||
// 构建返回报文， 固定报头
|
||||
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
|
||||
// 构建返回报文， 可变报头
|
||||
MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
|
||||
MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
|
||||
log.info("back--" + mqttMessageBack.toString());
|
||||
channel.writeAndFlush(mqttMessageBack);
|
||||
}
|
||||
|
||||
/**
|
||||
* 订阅确认
|
||||
*
|
||||
* @param channel
|
||||
* @param mqttMessage
|
||||
*/
|
||||
public static void suback(Channel channel, MqttMessage mqttMessage) {
|
||||
log.info("接收到订阅确认信息：{}", mqttMessage.toString());
|
||||
MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
|
||||
MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
|
||||
// 构建返回报文， 可变报头
|
||||
MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
|
||||
Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
|
||||
//log.info(topics.toString());
|
||||
List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
|
||||
for (int i = 0; i < topics.size(); i++) {
|
||||
grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
|
||||
}
|
||||
// 构建返回报文 有效负载
|
||||
MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
|
||||
// 构建返回报文 固定报头
|
||||
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2 + topics.size());
|
||||
// 构建返回报文 订阅确认
|
||||
MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
|
||||
log.info("回复订阅确认信息：{}", subAck.toString());
|
||||
channel.writeAndFlush(subAck);
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消订阅确认
|
||||
*
|
||||
* @param channel
|
||||
* @param mqttMessage
|
||||
*/
|
||||
public static void unsuback(Channel channel, MqttMessage mqttMessage) {
|
||||
MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
|
||||
// 构建返回报文 可变报头
|
||||
MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
|
||||
// 构建返回报文 固定报头
|
||||
MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
|
||||
// 构建返回报文 取消订阅确认
|
||||
MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
|
||||
log.info("back--" + unSubAck.toString());
|
||||
channel.writeAndFlush(unSubAck);
|
||||
}
|
||||
|
||||
/**
|
||||
* 心跳响应
|
||||
*
|
||||
* @param channel
|
||||
* @param mqttMessage
|
||||
*/
|
||||
public static void pingresp(Channel channel, MqttMessage mqttMessage) {
|
||||
// 心跳响应报文 11010000 00000000 固定报文
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
|
||||
MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
|
||||
log.info("MqttServer回复心跳消息：{}", mqttMessageBack.toString());
|
||||
channel.writeAndFlush(mqttMessageBack);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package com.jsowell.netty.server.mqtt;
|
||||
|
||||
import com.jsowell.common.constant.Constants;
|
||||
|
||||
import io.netty.bootstrap.ServerBootstrap;
|
||||
import io.netty.buffer.PooledByteBufAllocator;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||
import io.netty.handler.codec.mqtt.MqttDecoder;
|
||||
import io.netty.handler.codec.mqtt.MqttEncoder;
|
||||
import io.netty.handler.timeout.IdleStateHandler;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.CommandLineRunner;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
@Order(value = 2)
|
||||
public class MqttSever implements CommandLineRunner {
|
||||
|
||||
@Override
|
||||
public void run(String... args) throws Exception {
|
||||
InetSocketAddress address = new InetSocketAddress(Constants.SOCKET_IP, 1883);
|
||||
this.start(address);
|
||||
}
|
||||
|
||||
public void start(InetSocketAddress address) {
|
||||
// log.info("========NettyServer.start order 1");
|
||||
//配置服务端的NIO线程组
|
||||
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
|
||||
EventLoopGroup workerGroup = new NioEventLoopGroup();
|
||||
|
||||
try {
|
||||
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 future = mqttBootstrap.bind(address.getPort()).sync();
|
||||
if(future.isSuccess()){
|
||||
log.info("MqttServer启动成功, 开始监听端口:{}", address.getPort());
|
||||
future.channel().closeFuture().sync();
|
||||
} else {
|
||||
log.error("MqttServer启动失败", future.cause());
|
||||
}
|
||||
|
||||
//关闭channel和块，直到它被关闭
|
||||
future.channel().closeFuture().sync();
|
||||
} catch (Exception e) {
|
||||
log.error("MqttServer.start error", e);
|
||||
bossGroup.shutdownGracefully();
|
||||
workerGroup.shutdownGracefully();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2,9 +2,7 @@ package com.jsowell.netty.server.yunkuaichong;
|
||||
|
||||
import com.jsowell.common.constant.Constants;
|
||||
import io.netty.bootstrap.ServerBootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelOption;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
@@ -19,11 +17,11 @@ import java.net.InetSocketAddress;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
@Order(value = 1)
|
||||
public class NettyServer implements CommandLineRunner {
|
||||
@Resource
|
||||
private NettyServerChannelInitializer nettyServerChannelInitializer;
|
||||
|
||||
@Order(value = 1)
|
||||
@Override
|
||||
public void run(String... args) throws Exception {
|
||||
InetSocketAddress address = new InetSocketAddress(Constants.SOCKET_IP, Constants.SOCKET_PORT);
|
||||
@@ -66,8 +64,41 @@ 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();
|
||||
} catch (Exception e) {
|
||||
log.error("NettyServer.start error", e);
|
||||
bossGroup.shutdownGracefully();
|
||||
|
||||
Reference in New Issue
Block a user