venue/smart_venue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

netty-修改请求框架，并且注册客户端信息落地，之后通过调用netty接口来进行后续操作。

Browse Source
This commit is contained in:
limqhz
2020-07-11 23:33:02 +08:00
parent 3703ee2844
commit 6f08b24538
29 changed files with 266 additions and 689 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
api/src/main/resources/config/application.yml
View File

@@ -1,6 +1,5 @@
server:
port: 8093
# port: 8000
debug: true
app:

2
doc/v1.1.sql
View File

@@ -1,2 +0,0 @@
alter table sv_venue_lesson add column type TINYINT(4) NOT NULL DEFAULT '0' COMMENT '课程类型：0普通课程 1公益课程';
alter table sv_member add column ban_type TINYINT(4) NOT NULL DEFAULT '0' COMMENT '用户公益课黑名单：0正常 1 公益课黑名单';

3
entity/src/main/java/com/enums/DeviceStatusEnum.java
View File

@@ -2,6 +2,9 @@ package com.enums;
public enum DeviceStatusEnum {
OFFLINE(0, "断开连接"),
/**
* 重连服务器无法主动发起
*/
RECONNECT(1, "正在重连"),
ONLINE(2, "已连接");
public Integer value;

2
entity/src/main/java/com/sv/entity/Device.java
View File

@@ -28,7 +28,7 @@ public class Device implements Serializable {
private String stream;
/**
* 设备状态 0未连接 1正在重连 2已连接
* 设备状态 0未连接 1正在重连（无用） 2已连接
*/
private Integer status;

63
netty-pad/src/main/java/com/sv/netty/config/ChannelTypeCache.java
View File

@@ -1,63 +0,0 @@
package com.sv.netty.config;
import io.netty.util.internal.PlatformDependent;
import java.util.concurrent.ConcurrentMap;
/**
* 通道类型全局存储
*
* @author Demon
* @since 09/01/2018
*/
public class ChannelTypeCache {
public static final Integer TCP = 1;
public static final Integer WEB_SOCKET = 2;
/**
* 存储用户和通道类型
* value: 1Tcp 2WebSocket
*/
public static final ConcurrentMap<Integer, Integer> channelTypes = PlatformDependent.newConcurrentHashMap();
/**
* 设置通道类型
*
* @param memberId 会员编号
* @param channelType 通道类型：1Tcp 2WebSocket
*/
public static void putChannelType(Integer memberId, Integer channelType) {
channelTypes.put(memberId, channelType);
}
/**
* 获取通道类型
*
* @param memberId 会员编号
* @return 通道类型：1Tcp 2WebSocket
*/
public static Integer getChannelType(Integer memberId) {
return channelTypes.get(memberId);
}
/**
* 判断是否存在用户
*
* @param memberId 会员编号
* @return 是否存在用户
*/
public static boolean contains(Integer memberId) {
return channelTypes.containsKey(memberId);
}
/**
* 删除通道类型
*
* @param memberId 会员编号
*/
public static void removeChannelType(Integer memberId) {
channelTypes.remove(memberId);
}
}

7
netty-pad/src/main/java/com/sv/netty/config/ClientChannelCache.java
View File

@@ -7,7 +7,8 @@ import java.util.concurrent.ConcurrentMap;
public class ClientChannelCache {
/**
* 存储用户和通道类型
* 存储设备通道类型
* 此处存储客户端的channel 信息key 为 【deviceId + venueId】
* value: 1Tcp 2WebSocket
*/
public static final ConcurrentMap<String, Channel> channelTypes = PlatformDependent.newConcurrentHashMap();
@@ -29,14 +30,14 @@ public class ClientChannelCache {
/**
* 判断通道是否存在
*/
public static boolean contains(Integer clientId) {
public static boolean contains(String clientId) {
return channelTypes.containsKey(clientId);
}
/**
* 移除通道
*/
public static void removeChannelType(Integer clientId) {
public static void removeChannelType(String clientId) {
channelTypes.remove(clientId);
}
}

23
netty-pad/src/main/java/com/sv/netty/controller/MessageControler.java
View File

@@ -1,33 +1,18 @@
package com.sv.netty.controller;
import com.sv.entity.face.FaceRecognizeResponse;
import com.sv.netty.dto.ResponseDTO;
import com.sv.netty.service.impl.TcpMessageHandlerAdapter;
import com.sv.service.oms.DeviceService;
import com.sv.service.oms.VenueService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
@RestController
public class MessageControler {
@Resource()
private TcpMessageHandlerAdapter messageService;
@Resource
private DeviceService deviceService;
@Resource
private VenueService venueService;
@RequestMapping("/message/send/t")
public ResponseDTO sendMessage1() {
FaceRecognizeResponse recognizeResponse = new FaceRecognizeResponse();
recognizeResponse.setPerson(new FaceRecognizeResponse.PersonBeanX());
recognizeResponse.getPerson().setId(49);
messageService.enter(deviceService.findById(1), recognizeResponse, venueService.findById(32));
// FaceRecognizeResponse recognizeResponse = new FaceRecognizeResponse();
// recognizeResponse.setPerson(new FaceRecognizeResponse.PersonBeanX());
// recognizeResponse.getPerson().setId(49);
// messageService.enter(deviceService.findById(1), recognizeResponse, venueService.findById(32));
return ResponseDTO.ok();
}

14
netty-pad/src/main/java/com/sv/netty/controller/QRCodeControler.java
View File

@@ -1,7 +1,9 @@
package com.sv.netty.controller;
import com.sv.netty.config.ClientChannelCache;
import com.sv.netty.config.ErrorCode;
import com.sv.netty.dto.ResponseDTO;
import io.netty.channel.Channel;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@@ -30,4 +32,16 @@ public class QRCodeControler {
}
}
/**
* 微信小程序扫码之后，通知对应的线程进入加载界面，然后随即开始判断逻辑
* @return
*/
@RequestMapping("/testChannel")
public ResponseDTO testChannel(@RequestParam String id) {
System.out.println(ClientChannelCache.channelTypes.size());
Channel channel = ClientChannelCache.getChannelType(id);
channel.writeAndFlush("你好呀！");
return ResponseDTO.ok();
}
}

1
netty-pad/src/main/java/com/sv/netty/netty/MessageEncoder.java
View File

@@ -1,7 +1,6 @@
package com.sv.netty.netty;
import com.sv.netty.config.Constant;
import com.sv.netty.netty.message.BaseDto;
import com.sv.netty.utils.JsonUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;

66
netty-pad/src/main/java/com/sv/netty/netty/ServerHandler.java
View File

@@ -3,12 +3,17 @@ package com.sv.netty.netty;
import com.sv.netty.config.Constant;
import com.sv.netty.config.SpringContextHolder;
import com.sv.netty.netty.message.ChannelParam;
import com.sv.netty.netty.message.HeartBeat;
import com.sv.netty.service.MessageService;
import com.sv.netty.utils.JsonUtils;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.util.Set;
/**
* 服务器接受TCP协议数据处理器
@@ -37,22 +42,30 @@ public class ServerHandler extends SimpleChannelInboundHandler<String> {
super.channelActive(ctx);
}
/**
* 客户端只会报心跳
* @param ctx
* @param msg
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
String clientIp = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getClientIp();
try {
messageService.receive(clientIp,ctx.channel(), msg);
HeartBeat hb = JsonUtils.decode(msg,HeartBeat.class);
logger.info("客户端" + hb.getDeviceName() + "&" + hb.getVenueId() + "【" + clientIp + "】上报心跳...");
ctx.channel().attr(Constant.CHANNEL_PARAM).get().setVenueId(hb.getVenueId());
ctx.channel().attr(Constant.CHANNEL_PARAM).get().setDeviceName(hb.getDeviceName());
messageService.online(clientIp,ctx.channel(), hb);
} catch (Exception e) {
logger.error("[" + clientIp + "] host unknown error");
logger.error("[" + clientIp + "] host unknown error",e);
}
}
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
public void channelRegistered(ChannelHandlerContext ctx) {
String clientIP = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();
logger.info("There is a client Registered. ip:" + clientIP);
ctx.channel().attr(Constant.CHANNEL_PARAM).set(new ChannelParam(clientIP));
messageService.online(ctx.channel());
}
@@ -62,22 +75,17 @@ public class ServerHandler extends SimpleChannelInboundHandler<String> {
}
/**
* Calls {@link ChannelHandlerContext#fireChannelInactive()} to forward
* to the next {@link ChannelHandler} in the {@link ChannelPipeline}.
* <p>
* Sub-classes may override this method to change behavior.
*
* 当通道失效就会触发
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
public void channelInactive(ChannelHandlerContext ctx) {
String clientIP = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();
logger.error("Client ip [" + clientIP + "] has inactive");
Integer memberId = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getMemberId();
String nickname = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getNickname();
String machineSn = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getDeviceSn();
messageService.doOffLine(memberId, nickname,machineSn);
messageService.destory(ctx.channel());
Integer venueId = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getVenueId();
String deviceName = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getDeviceName();
messageService.Offline(deviceName,venueId);
}
/**
@@ -90,6 +98,9 @@ public class ServerHandler extends SimpleChannelInboundHandler<String> {
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
logger.error("ServerHandler exceptionCaught",cause);
Channel channel = ctx.channel();
Integer venueId = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getVenueId();
String deviceName = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getDeviceName();
messageService.Offline(deviceName,venueId);
if(channel.isActive()) {
// 错误产生，关闭连接
ctx.close();
@@ -101,16 +112,19 @@ public class ServerHandler extends SimpleChannelInboundHandler<String> {
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
// Do nothing 客户端这个方法用来发心跳，
// 可以考虑再ctx 获取客户端的上下文信息，将该客户移除我们的操作 TODO 值得尝试主要是要区分客户端
// IdleStateEvent event = (IdleStateEvent) evt;
// // 如果Channel读取数据闲置，则关闭连接
// if (event.state() == IdleState.READER_IDLE) {
// String clientIP = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();
// logger.info("Client [" + clientIP + "] has idle");
// messageService.destory(ctx.channel());
// ctx.channel().close();
// }
// 服务器统计现在有的客户端数量， 客户端这个方法用来发心跳
if (evt instanceof IdleStateEvent){
IdleState state = ((IdleStateEvent) evt).state();
if (state == IdleState.READER_IDLE){
logger.info("IdleStateEvent READER_IDLE 超时");
Integer venueId = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getVenueId();
String deviceName = ctx.channel().attr(Constant.CHANNEL_PARAM).get().getDeviceName();
messageService.Offline(deviceName,venueId);
ctx.channel().close();
}
}
Set<String> conns = messageService.countConnection();
logger.info("count connected device ! the count is " + conns.size() + " and they are + [" + conns.toString() + "]" );
}
}

12
netty-pad/src/main/java/com/sv/netty/netty/ServerProtocolInitializer.java
View File

@@ -20,23 +20,21 @@ import org.springframework.stereotype.Component;
@Component
public class ServerProtocolInitializer extends ChannelInitializer<SocketChannel> {
private final int IDLE_TIME = 30; //连接检测空闲时间
private final int READ_TIME = 30; //读超时时间
private final int WRITE_TIME = 30; //写超时时间
private final int READ_TIMEOUT = 60; //读超时时间
private final int IDLE_TIME = 120; //连接检测空闲时间
// private final int READ_TIMEOUT = 30; //读超时时间
@Override
protected void initChannel(SocketChannel ch){
ChannelPipeline pipeline = ch.pipeline();
// 超时设置
pipeline.addLast(new ReadTimeoutHandler(READ_TIMEOUT));
// 超时设置（如果客户端频繁有人操作，则不会报心跳，这样就会触发超时，关闭链接）
// pipeline.addLast(new ReadTimeoutHandler(READ_TIMEOUT));
// 通过指定的长度来标识整包的信息，这样就可以自动的处理粘包和半包的问题
pipeline.addLast(new DelimiterBasedFrameDecoder(2048,
Unpooled.wrappedBuffer(Constant.DELIMITER_WORD.getBytes())));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new MessageEncoder());
// 心跳检测机制，通过调用触发下一个handler userEventTriggered 方法
pipeline.addLast(new IdleStateHandler(READ_TIME, WRITE_TIME,IDLE_TIME));
pipeline.addLast(new IdleStateHandler(IDLE_TIME, IDLE_TIME,IDLE_TIME));
pipeline.addLast(new ServerHandler());
}

93
netty-pad/src/main/java/com/sv/netty/netty/message/BaseDto.java
View File

@@ -1,93 +0,0 @@
package com.sv.netty.netty.message;
import com.sv.entity.Device;
import java.io.Serializable;
/**
* Created by peakren on 19/01/2018.
*/
public class BaseDto implements Serializable {
private static final long serialVersionUID = 3139438146199448677L;
private String cmdId; //指令
private String deviceId; //设备ID
/**
* 客户端IP
*/
private String clientIp;
private String content;
/**
* 状态码 0正常 1错误
*/
private int errorCode = 0;
/**
* 错误返回信息
*/
private String errorMsg;
public String getCmdId() {
return cmdId;
}
public void setCmdId(String cmdId) {
this.cmdId = cmdId;
}
public String getDeviceId() {
return deviceId;
}
public void setDeviceId(String deviceId) {
this.deviceId = deviceId;
}
/**
* 客户端IP,获取的内网IP
*
* @return
*/
public String getClientIp() {
return clientIp;
}
public void setClientIp(String clientIp) {
this.clientIp = clientIp;
}
public int getErrorCode() {
return errorCode;
}
public void setErrorCode(int errorCode) {
this.errorCode = errorCode;
}
public String getErrorMsg() {
return errorMsg;
}
public void setErrorMsg(String errorMsg) {
this.errorMsg = errorMsg;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}

61
netty-pad/src/main/java/com/sv/netty/netty/message/ChannelParam.java
View File

@@ -16,34 +16,15 @@ public class ChannelParam {
/**
* 设备编号
*/
private String deviceSn;
private String deviceName;
/**
* 会员编号
* 设备所在场馆
*/
private Integer memberId;
private Integer venueId;
/**
* 昵称
*/
private String nickname;
/**
* 头像
*/
private String avatar;
public ChannelParam() {
}
public ChannelParam(String clientIp){
this.clientIp = clientIp;
}
public ChannelParam(String clientIp, String deviceSn) {
this.clientIp = clientIp;
this.deviceSn = deviceSn;
public ChannelParam(String clientIP) {
this.clientIp = clientIP;
}
public String getClientIp() {
@@ -54,35 +35,19 @@ public class ChannelParam {
this.clientIp = clientIp;
}
public String getDeviceSn() {
return deviceSn;
public String getDeviceName() {
return deviceName;
}
public void setDeviceSn(String deviceSn) {
this.deviceSn = deviceSn;
public void setDeviceName(String deviceName) {
this.deviceName = deviceName;
}
public Integer getMemberId() {
return memberId;
public Integer getVenueId() {
return venueId;
}
public void setMemberId(Integer memberId) {
this.memberId = memberId;
}
public String getNickname() {
return nickname;
}
public void setNickname(String nickname) {
this.nickname = nickname;
}
public String getAvatar() {
return avatar;
}
public void setAvatar(String avatar) {
this.avatar = avatar;
public void setVenueId(Integer venueId) {
this.venueId = venueId;
}
}

39
netty-pad/src/main/java/com/sv/netty/netty/message/Cmd.java
View File

@@ -1,39 +0,0 @@
package com.sv.netty.netty.message;
/**
* 消息协议指令定义
*
* @Author peakren
* @Date 08/12/2017 11:51 AM
*/
public enum Cmd {
HB("hb", "心跳"),
FACEID("faceid", "人脸识别"),
OPEN_DOOR("open_door", "开门"),
FACEID_RESPONSE("faceid_response_upload", "人脸结果");
public String id;
public String text;
Cmd(String id, String text) {
this.id = id;
this.text = text;
}
public static Cmd getCmd(String id) {
for (Cmd cmd : Cmd.values()) {
if (cmd.id.equalsIgnoreCase(id)) {
return cmd;
}
}
return Cmd.HB;
}
}

36
netty-pad/src/main/java/com/sv/netty/netty/message/HeartBeat.java Normal file
View File

@@ -0,0 +1,36 @@
package com.sv.netty.netty.message;
import com.google.gson.annotations.Expose;
import java.io.Serializable;
/**
* 客户端心跳数据包
* HeartBeat.java
*
* @author peakren
* @date 07/12/2017 10:23 PM
*/
public class HeartBeat implements Serializable {
private Integer venueId; //场馆号
private String deviceName; //设备号
public Integer getVenueId() {
return venueId;
}
public void setVenueId(Integer venueId) {
this.venueId = venueId;
}
public String getDeviceName() {
return deviceName;
}
public void setDeviceName(String deviceName) {
this.deviceName = deviceName;
}
}

23
netty-pad/src/main/java/com/sv/netty/netty/message/MessageDto.java
View File

@@ -1,23 +0,0 @@
package com.sv.netty.netty.message;
public class MessageDto extends BaseDto {
private Object result;
private Integer door;
public Object getResult() {
return result;
}
public void setResult(Object result) {
this.result = result;
}
public Integer getDoor() {
return door;
}
public void setDoor(Integer door) {
this.door = door;
}
}

28
netty-pad/src/main/java/com/sv/netty/service/MessageService.java
View File

@@ -1,8 +1,10 @@
package com.sv.netty.service;
import com.sv.netty.dto.MemberDto;
import com.sv.netty.netty.message.HeartBeat;
import io.netty.channel.Channel;
import java.util.Set;
/**
* 消息实现接口
*
@@ -12,24 +14,24 @@ import io.netty.channel.Channel;
public interface MessageService {
/**
* 接受设备消息
* 接受设备心跳，设备上线
*
* @param channel
* @param content
* @param heartBeat
*/
void receive(String clientId, Channel channel, String content);
void online(Channel channel);
void online(String clientId, Channel channel, HeartBeat heartBeat);
/**
* 销毁设备的连接
* 设备下线
*
* @param channel
* @param deviceName
* @param venueId
*/
void destory(Channel channel);
void Offline(String deviceName, Integer venueId);
void doOffLine(Integer memberId, String nickname, String machineSn);
void sendMessage(MemberDto memberDto);
/**
* 统计目前的链接数
* @return
*/
Set<String> countConnection();
}

172
netty-pad/src/main/java/com/sv/netty/service/impl/TcpMessageHandlerAdapter.java
View File

@@ -1,19 +1,18 @@
package com.sv.netty.service.impl;
import com.enums.FaceRecognizeEnum;
import com.sv.dto.EnterResult;
import com.sv.entity.Device;
import com.sv.entity.Member;
import com.sv.entity.MemberEnterVenueLog;
import com.sv.entity.Venue;
import com.sv.entity.face.FaceRecognizeResponse;
import com.sv.netty.config.ClientChannelCache;
import com.sv.netty.dto.MemberDto;
import com.sv.netty.netty.message.BaseDto;
import com.sv.netty.netty.message.Cmd;
import com.sv.netty.netty.message.MessageDto;
import com.sv.netty.netty.message.HeartBeat;
import com.sv.netty.netty.message.Result;
import com.sv.netty.service.MessageService;
import com.sv.netty.utils.JsonMapper;
import com.sv.netty.utils.JsonUtils;
import com.sv.service.api.MemberEnterVenueLogService;
import com.sv.service.api.MemberService;
import com.sv.service.api.VenueService;
@@ -23,7 +22,6 @@ import com.ydd.oms.entity.sys.Config;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
@@ -47,27 +45,18 @@ public class TcpMessageHandlerAdapter implements MessageService {
private Logger logger = LoggerFactory.getLogger(TcpMessageHandlerAdapter.class);
//
// @Value("${nettym.url}")
// private String url;
@Resource
private MemberService memberService;
@Resource
private VenueService venueService;
private Set<Channel> channels = new HashSet<>();
@Resource
private DeviceService deviceService;
@Resource
private ConfigService configService;
private Channel enter;
private Channel out;
@Resource(name = "scheduledExecutorService")
private ScheduledExecutorService scheduledExecutorService;
@@ -79,82 +68,60 @@ public class TcpMessageHandlerAdapter implements MessageService {
private RestTemplate restTemplate;
public void sendMessage(MemberDto memberDto, Integer deviceId) {
MessageDto messageDto = new MessageDto();
messageDto.setCmdId(Cmd.FACEID.id);
messageDto.setResult(memberDto);
logger.info("发送消息" + channels.size() + "；deviceId:" + deviceId);
if (deviceId == 1) {
if (enter != null) {
enter.writeAndFlush(messageDto);
}
} else {
out.writeAndFlush(messageDto);
}
// MessageDto messageDto = new MessageDto();
// messageDto.setCmdId(Cmd.FACEID.id);
// messageDto.setResult(memberDto);
// logger.info("发送消息" + channels.size() + "；deviceId:" + deviceId);
// if (deviceId == 1) {
// if (enter != null) {
// enter.writeAndFlush(messageDto);
// }
// } else {
// out.writeAndFlush(messageDto);
// }
}
public void sendOpenMessage(MessageDto messageDto, Integer deviceId) {
logger.info("发送消息开门" + channels.size());
if (enter != null) {
enter.writeAndFlush(messageDto);
/**
* 处理心跳信息，存储心跳信息
* @param clientId
* @param channel
* @param heartBeat
*/
@Override
public void online(String clientId, Channel channel, HeartBeat heartBeat) {
// 处理心跳信息
if (!ClientChannelCache.contains(heartBeat.getDeviceName()+heartBeat.getVenueId())){
// 此处存储客户端的channel 信息key 为 deviceId + venueId
Venue thisVenue = venueService.findById(heartBeat.getVenueId());
if (thisVenue == null ){
logger.error("this client choose venue Error! venueId == " + heartBeat.getVenueId());
} else {
ClientChannelCache.putChannelType(heartBeat.getDeviceName()+heartBeat.getVenueId(),channel);
deviceService.online(heartBeat.getDeviceName(),heartBeat.getVenueId(),thisVenue.getType(),clientId);
}
}
}
/**
* 接收设备发送过来的协议数据,业务处理后并返回
*
* @param channel netty的连接通道会话
* @param content 消息内容
* 设备下线
* @param deviceName
* @param venueId
*/
@Override
public void receive(String clientId, Channel channel, String content) {
try {
//解析数据
Cmd cmd = resolveCmd(content);
logger.info("收到[" + clientId + "]消息" + cmd.text);
switch (cmd) {
case HB:
break;
case FACEID_RESPONSE:
synchronized (this) {
resolve(channel, content);
}
break;
default:
break;
}
} catch (Exception e) {
logger.error("消息内容：" + content);
logger.error("处理TCP消息异常", e);
public void Offline(String deviceName, Integer venueId) {
if (deviceName != null && venueId != null){
ClientChannelCache.removeChannelType(deviceName+ venueId);
deviceService.offline(deviceName,venueId);
}
}
/**
* 统计目前有多少链接数
* @return
*/
@Override
public void online(Channel channel) {
channels.add(channel);
}
public synchronized void resolve(Channel channel, String content) {
BaseDto baseDto = JsonMapper.nonEmptyMapper().fromJson(content, BaseDto.class);
FaceRecognizeResponse response = JsonMapper.nonNullMapper().fromJson(baseDto.getContent(), FaceRecognizeResponse.class);
Device device = deviceService.findById(Integer.parseInt(baseDto.getDeviceId()));
Venue venue = venueService.findById(device.getVenueId());
if (FaceRecognizeEnum.RECOGNIZED.name.equals(response.getType())) {
//识别成功
if (response.getPerson() != null) {
if (device.getId() == 1) {
//进门
logger.info("开门");
enter = channel;
enter(device, response, venue);
} else if (device.getId() == 2) {
//出门
logger.info("出门");
out = channel;
out(device, response, venue);
}
}
}
public Set<String> countConnection() {
return ClientChannelCache.channelTypes.keySet();
}
public void enter(Device device, FaceRecognizeResponse response, Venue venue) {
@@ -221,18 +188,18 @@ public class TcpMessageHandlerAdapter implements MessageService {
scheduledExecutorService.schedule(new Runnable() {
@Override
public void run() {
MessageDto messageDto = new MessageDto();
messageDto.setCmdId(Cmd.OPEN_DOOR.id);
messageDto.setDoor(1);
sendOpenMessage(messageDto, 1);
// MessageDto messageDto = new MessageDto();
// messageDto.setCmdId(Cmd.OPEN_DOOR.id);
// messageDto.setDoor(1);
// sendOpenMessage(messageDto, 1);
}
}, 5, TimeUnit.SECONDS);
} else {
//开门
MessageDto messageDto = new MessageDto();
messageDto.setCmdId(Cmd.OPEN_DOOR.id);
messageDto.setDoor(1);
sendOpenMessage(messageDto, 1);
// MessageDto messageDto = new MessageDto();
// messageDto.setCmdId(Cmd.OPEN_DOOR.id);
// messageDto.setDoor(1);
// sendOpenMessage(messageDto, 1);
logger.info(member.getId() + "入场成功:" + member.getMoney().toString());
}
@@ -280,10 +247,10 @@ public class TcpMessageHandlerAdapter implements MessageService {
memberMessageDto.setPlaceName("");
memberMessageDto.setMessage("欢迎下次再来" + venue.getName());
memberMessageDto.setCode(2);
MessageDto messageDto = new MessageDto();
messageDto.setCmdId(Cmd.OPEN_DOOR.id);
messageDto.setDoor(2);
sendOpenMessage(messageDto, device.getId());
// MessageDto messageDto = new MessageDto();
// messageDto.setCmdId(Cmd.OPEN_DOOR.id);
// messageDto.setDoor(2);
// sendOpenMessage(messageDto, device.getId());
sendMessage(memberMessageDto, device.getId());
venueService.addNumber(venue.getId(), -1, member.getId());
venue = venueService.findById(venue.getId());
@@ -306,27 +273,4 @@ public class TcpMessageHandlerAdapter implements MessageService {
logger.info("灯光结果" + JsonMapper.nonDefaultMapper().toJson(result));
}
@Override
public void destory(Channel channel) {
channels.remove(channel);
}
@Override
public void doOffLine(Integer memberId, String nickname, String machineSn) {
}
@Override
public void sendMessage(MemberDto memberDto) {
}
protected Cmd resolveCmd(String message) {
BaseDto baseDto = JsonMapper.nonEmptyMapper().fromJson(message, BaseDto.class);
return Cmd.getCmd(baseDto.getCmdId());
}
}

4
netty-pad/src/main/resources/config/application.yml
View File

@@ -78,10 +78,6 @@ face:
account: test@test.com
pwd: 123456
nettym:
url: http://127.0.0.1:8021/netty/message/send
#netty服务器配置
netty:
port: 56791

15
netty-pad/src/test/com/test/netty/client/ClientHandler.java
View File

@@ -1,8 +1,6 @@
package com.test.netty.client;
import com.test.netty.client.message.Cmd;
import com.test.netty.client.message.HeartBeat;
import com.test.netty.client.message.Message;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
@@ -79,13 +77,11 @@ public class ClientHandler extends SimpleChannelInboundHandler<String> {
* 封装心跳请求包
* @throws Exception
*/
private Message getHbMessage() {
private HeartBeat getHbMessage() {
HeartBeat hb = new HeartBeat();
hb.setVersionCode("AppUtil.getVersionCode(StartApplication.getAppContext())");
Message message = new Message();
message.setCmdId(Cmd.HB.id);
message.setDeviceId("DeviceIdUtil.generateDeviceId(mContext)");
return message;
hb.setDeviceName("DeviceIdUtil.generateDeviceId(mContext)");
hb.setVenueId(32); // TODO
return hb;
}
/**
@@ -95,8 +91,7 @@ public class ClientHandler extends SimpleChannelInboundHandler<String> {
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
logger.error("ClientHandler exceptionCaught",cause);
Channel channel = ctx.channel();
if(channel.isActive()) {

2
netty-pad/src/test/com/test/netty/client/ClientInitializer.java
View File

@@ -14,8 +14,6 @@ import io.netty.handler.timeout.ReadTimeoutHandler;
public class ClientInitializer extends ChannelInitializer<SocketChannel> {
private final static int TIME_HEART_BEAT = 20;
private final int READ_TIMEOUT = 60; //读超时时间
public ClientThread.ReConnectHandler reConnectHandler;
public ClientHandler dmClientHandler;

1
netty-pad/src/test/com/test/netty/client/MessageEncoder.java
View File

@@ -5,7 +5,6 @@ import com.sv.netty.utils.JsonUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import com.test.netty.client.message.Message;
import java.nio.charset.Charset;

86
netty-pad/src/test/com/test/netty/client/message/BaseDto.java
View File

@@ -1,86 +0,0 @@
package com.test.netty.client.message;
import com.google.gson.annotations.Expose;
import java.io.Serializable;
/**
* Created by peakren on 19/01/2018.
*/
public class BaseDto implements Serializable {
private static final long serialVersionUID = 3139438146199448677L;
@Expose
private String cmdId; //指令
@Expose
private String deviceId; //设备ID
/**
* 客户端IP
*/
@Expose
private String clientIp;
/**
* 状态码 0正常 1错误
*/
@Expose
private int errorCode = 0;
/**
* 错误返回信息
*/
@Expose
private String errorMsg;
public String getCmdId() {
return cmdId;
}
public void setCmdId(String cmdId) {
this.cmdId = cmdId;
}
public String getDeviceId() {
return deviceId;
}
public void setDeviceId(String deviceId) {
this.deviceId = deviceId;
}
/**
* 客户端IP,获取的内网IP
*
* @return
*/
public String getClientIp() {
return clientIp;
}
public void setClientIp(String clientIp) {
this.clientIp = clientIp;
}
public int getErrorCode() {
return errorCode;
}
public void setErrorCode(int errorCode) {
this.errorCode = errorCode;
}
public String getErrorMsg() {
return errorMsg;
}
public void setErrorMsg(String errorMsg) {
this.errorMsg = errorMsg;
}
}

45
netty-pad/src/test/com/test/netty/client/message/Cmd.java
View File

@@ -1,45 +0,0 @@
package com.test.netty.client.message;
/**
* 消息协议指令定义
*
* @Author peakren
* @Date 08/12/2017 11:51 AM
*/
public enum Cmd {
HB("hb", "心跳"),
FACEID("faceid", "人脸识别"),
FACEID_RESPONSE("faceid_response_upload", "识别结果"),
RECEV_FACE_IMAGE("recev_face_image", "接收人脸照片"),
RECEV_FACE_IMAGE_R("recev_face_image_r", "返回上传图片结果"),
OPEN_DOOR("open_door", "开门禁");
public String id;
public String text;
Cmd(String id, String text) {
this.id = id;
this.text = text;
}
public static Cmd getCmd(String id) {
for (Cmd cmd : Cmd.values()) {
if (cmd.id.equalsIgnoreCase(id)) {
return cmd;
}
}
return Cmd.HB;
}
}

26
netty-pad/src/test/com/test/netty/client/message/HeartBeat.java
View File

@@ -15,30 +15,24 @@ import java.io.Serializable;
public class HeartBeat implements Serializable {
@Expose
private String versionCode; //客户端版本号
private Integer venueId; //场馆号
@Expose
private String apkUrl; //升级的软件下载地址
private String deviceName; //设备号
public String getVersionCode() {
return versionCode;
public Integer getVenueId() {
return venueId;
}
/**
* 客户端版本号
*
* @param versionCode
*/
public void setVersionCode(String versionCode) {
this.versionCode = versionCode;
public void setVenueId(Integer venueId) {
this.venueId = venueId;
}
public String getApkUrl() {
return apkUrl;
public String getDeviceName() {
return deviceName;
}
public void setApkUrl(String apkUrl) {
this.apkUrl = apkUrl;
public void setDeviceName(String deviceName) {
this.deviceName = deviceName;
}
}

55
netty-pad/src/test/com/test/netty/client/message/Message.java
View File

@@ -1,55 +0,0 @@
package com.test.netty.client.message;
import com.google.gson.annotations.Expose;
/**
* <pre>上位机与通讯服务器的数据协议格式</pre>
*
* @author peakren
* @date 07/12/2017 9:57 PM
*/
public class Message extends BaseDto {
private static final long serialVersionUID = -7944124768291562453L;
/**
* 消息内容
*/
@Expose
private MemberDto result;
/**
* 机器识别返回字符串
*/
@Expose
private String content;
public int getDoor() {
return door;
}
public void setDoor(int door) {
this.door = door;
}
@Expose
private int door;
public MemberDto getResult() {
return result;
}
public void setResult(MemberDto result) {
this.result = result;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}

14
service/src/main/java/com/sv/mapper/DeviceMapper.java
View File

@@ -62,4 +62,18 @@ public interface DeviceMapper {
List<DeviceDTO> findAllDTO();
Integer countByStream(@Param("stream") String stream,@Param("id") Integer id);
/**
* 更新设备状态为 0 - 未连接
*/
void offline(@Param("venueId") Integer venueId,@Param("deviceName") String deviceName);
/**
* 更新设备状态为 2 - 连接成功
*/
void online(@Param("venueId") Integer venueId,@Param("deviceName") String deviceName);
Integer checkDevice(@Param("deviceName") String deviceName,@Param("venueId") Integer venueId);
}

33
service/src/main/java/com/sv/service/oms/DeviceService.java
View File

@@ -20,6 +20,8 @@ import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;
import javax.annotation.Resource;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
/**
@@ -29,7 +31,6 @@ import java.util.List;
* @since 2018-08-09
*/
@Service
@Transactional(readOnly = true)
public class DeviceService extends BaseServiceImpl {
private final Logger logger = LoggerFactory.getLogger(DeviceService.class);
@@ -131,26 +132,30 @@ public class DeviceService extends BaseServiceImpl {
/**
* 设备连接断开
* @param deviceId
*/
@Transactional
public void offline(Integer deviceId){
Device device = new Device();
device.setId(deviceId);
device.setStatus(DeviceStatusEnum.OFFLINE.value);
deviceMapper.update(device);
public void offline(String deviceName, Integer venueId){
deviceMapper.offline(venueId,deviceName);
}
/**
* 设备连接成功
* @param deviceId
* 新的设备注册的逻辑
*/
@Transactional
public void online(Integer deviceId){
Device device = new Device();
device.setId(deviceId);
device.setStatus(DeviceStatusEnum.ONLINE.value);
deviceMapper.update(device);
public void online(String deviceName,Integer venueId,Integer venueType,String deviceIp){
Device device = new Device();
device.setVenueId(venueId);
device.setVenueType(venueType);
device.setName(deviceName);
device.setStatus(DeviceStatusEnum.ONLINE.value);
device.setStream(deviceIp);
if(deviceMapper.checkDevice(deviceName,venueId) > 0){
logger.info(deviceName + venueId + "设备已存在，该设备。");
deviceMapper.online(venueId, deviceName);
}else {
logger.info("落地客户端信息clientId = " + deviceIp + "&deviceName = " + deviceName + "&venueId = " + venueId);
deviceMapper.insert(device);
}
}
/**

28
service/src/main/resources/mybatis/mapper/sv/DeviceMapper.xml
View File

@@ -8,7 +8,7 @@
<result column="stream" property="stream"/>
<result column="venue_id" property="venueId"/>
<result column="venue_type" property="venueType"/>
<result column="status" property="status"></result>
<result column="status" property="status"/>
<result column="created_id" property="createdId"/>
<result column="modified_id" property="modifiedId"/>
<result column="created_time" property="createdTime"/>
@@ -63,7 +63,7 @@
</select>
<!-- 新增-->
<insert id="insert" parameterType="com.sv.entity.Device" useGeneratedKeys="true" keyProperty="id">
<insert id="insert" parameterType="com.sv.entity.Device">
INSERT INTO
<include refid="tableName"></include>
@@ -81,7 +81,10 @@
venue_id,
</if>
<if test="venueType != null">
venue_type
venue_type,
</if>
<if test="status != null">
status,
</if>
<if test="createdId != null">
created_id,
@@ -115,6 +118,9 @@
<if test="venueType != null">
#{venueType},
</if>
<if test="status != null">
#{status},
</if>
<if test="createdId != null">
#{createdId},
</if>
@@ -220,4 +226,20 @@
</if>
</select>
<update id="offline">
UPDATE sv_device set status = 0,modified_time = now()
WHERE name = #{deviceName} and venue_id=#{venueId}
</update>
<update id="online">
UPDATE sv_device set status = 2,modified_time = now()
WHERE name = #{deviceName} and venue_id=#{venueId}
</update>
<select id="checkDevice" resultType="java.lang.Integer">
SELECT count(1) FROM sv_device
WHERE name = #{deviceName} and venue_id=#{venueId} and deleted = 0
</select>
</mapper>