在SpringBoot项目中需要对接三方系统，对接协议是TCP，需实现一个TCP客户端接收
服务端发送的数据并按照16进制进行解析数据，然后对数据进行过滤，将指定类型的数据
通过mybatis存储进mysql数据库中。并且当tcp服务端断连时，tcp客户端能定时检测并发起重连。
全流程效果
注：
博客：
霸道流氓气质的博客_CSDN博客-C#,架构之路,SpringBoot领域博主
1、SpringBoot+Netty实现TCP客户端
本篇参考如下博客，在如下博客基础上进行修改
Springboot+Netty搭建基于TCP协议的客户端（二）:
https://www.cnblogs.com/haolb123/p/16553005.html
上面博客提供的示例代码
https://download.csdn.net/download/myyhtw/12369531
引入Netty的依赖
<!-- netty依赖 springboot2.x自动导入版本-->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>2、新建Netty的client类
package com.badao.demo.netty;
import com.badao.demo.global.Global;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
*
* netty 客户端
*
*/
public class BootNettyClient {
public void connect(int port, String host) throws Exception{
/**
* 客户端的NIO线程组
*
*/
EventLoopGroup group = new NioEventLoopGroup();
try {
/**
* Bootstrap 是一个启动NIO服务的辅助启动类 客户端的
*/
Bootstrap bootstrap = new Bootstrap();
bootstrap = bootstrap.group(group);
bootstrap = bootstrap.channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true);
/**
* 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息
*/
bootstrap = bootstrap.handler(new BootNettyChannelInitializer<SocketChannel>());
/**
* 连接服务端
*/
ChannelFuture future = bootstrap.connect(host, port).sync();
if(future.isSuccess()) {
//是否连接tcp成功
Global.getInstance().canTcpConnected = true;
Channel channel = future.channel();
String id = future.channel().id().toString();
BootNettyClientChannel bootNettyClientChannel = new BootNettyClientChannel();
bootNettyClientChannel.setChannel(channel);
bootNettyClientChannel.setCode("clientId:"+id);
BootNettyClientChannelCache.save("clientId:"+id, bootNettyClientChannel);
System.out.println("netty client start success="+id);
/**
* 等待连接端口关闭
*/
future.channel().closeFuture().sync();
}else{
}
} finally {
/**
* 退出，释放资源
*/
group.shutdownGracefully().sync();
}
}
}注意这里的在连接成功之后的修改
新增了一个全局的单例变量类Global，用来作为断线重连的判断，后面后具体代码实现。
接着将clientId保存的实现，可以根据自己需要决定是否保留，不需要可删除，并且下面
第4条BootNettyClientChannel以及第5条BootNettyClientChannelCache也可做相应的删除或修改。
3、新建通道初始化
package com.badao.demo.netty;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
/**
* 通道初始化
*
*/
@ChannelHandler.Sharable
public class BootNettyChannelInitializer<SocketChannel> extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) {
/**
* 自定义ChannelInboundHandlerAdapter
*/
ch.pipeline().addLast(new BootNettyChannelInboundHandlerAdapter());
}
}注意与原有不一样的是这里删掉了自定义解码器的实现
这里根据自己实际情况决定是否保留以及格式，否则会提示
String cannot be cast to io.netty.buffer.ByteBuf
4、新建通道对象
package com.badao.demo.netty;
import io.netty.channel.Channel;
public class BootNettyClientChannel {
// 连接客户端唯一的code
private String code;
// 客户端最新发送的消息内容
private String last_data;
private transient volatile Channel channel;
public String getCode() {
return code;
}
public void setCode(String code) {
this.code = code;
}
public Channel getChannel() {
return channel;
}
public void setChannel(Channel channel) {
this.channel = channel;
}
public String getLast_data() {
return last_data;
}
public void setLast_data(String last_data) {
this.last_data = last_data;
}
}5、新建保存ClientChannel的Cache类
package com.badao.demo.netty;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class BootNettyClientChannelCache {
public static volatile Map<String, BootNettyClientChannel> channelMapCache = new ConcurrentHashMap<String, BootNettyClientChannel>();
public static void add(String code, BootNettyClientChannel channel){
channelMapCache.put(code,channel);
}
public static BootNettyClientChannel get(String code){
return channelMapCache.get(code);
}
public static void remove(String code){
channelMapCache.remove(code);
}
public static void save(String code, BootNettyClientChannel channel) {
if(channelMapCache.get(code) == null) {
add(code,channel);
}
}
}6、最重要的是新建客户端I/O数据读写处理类
package com.badao.demo.netty;
import com.badao.demo.entity.BusStallProptection;
import com.badao.demo.entity.StallVo;
import com.badao.demo.global.Global;
import com.badao.demo.mapper.BusStallProptectionMapper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Date;
/**
*
* I/O数据读写处理类
*
*/
@ChannelHandler.Sharable
@Component
public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
public static BootNettyChannelInboundHandlerAdapter bootNettyChannelInboundHandlerAdapter;
//1.正常注入[记得主类也需要使用@Component注解]
@Autowired
BusStallProptectionMapper busStallProptectionMapper;
//2.初始化构造方法一定要有
public BootNettyChannelInboundHandlerAdapter(){
}
//3.容器初始化的时候进行执行-这里是重点
@PostConstruct
public void init() {
bootNettyChannelInboundHandlerAdapter = this;
bootNettyChannelInboundHandlerAdapter.busStallProptectionMapper = this.busStallProptectionMapper;
}
/**
* 从服务端收到新的数据时，这个方法会在收到消息时被调用
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if(msg == null){
return;
}
ByteBuf buf = (ByteBuf) msg;
byte[] bytes = new byte[buf.readableBytes()];
// 复制内容到字节数组bytes
buf.readBytes(bytes);
// 将接收到的数据转为字符串，此字符串就是客户端发送的字符串
String receiveStr = NettyConnectHelper.receiveHexToString(bytes);
StallVo stallVo = NettyConnectHelper.receiveHexToObj(bytes);
BootNettyClientChannel bootNettyClientChannel = BootNettyClientChannelCache.get("clientId:"+ctx.channel().id().toString());
if(bootNettyClientChannel != null){
//判断指定状态的数据进行处理
if(Global.getInstance().abnormalCarStatusList.contains(stallVo.getCarStatus())){
BusStallProptection busStallProptection = BusStallProptection.builder()
.carNumber(stallVo.getCarNumber())
.carState(stallVo.getCarStatus())
.stallScope(stallVo.getAreaNumber())
.rawData(receiveStr)
.uploadTime(new Date())
.build();
//插入数据库
bootNettyChannelInboundHandlerAdapter.busStallProptectionMapper.insert(busStallProptection);
}
bootNettyClientChannel.setLast_data(msg.toString());
}
}
/**
* 从服务端收到新的数据、读取完成时调用
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
ctx.flush();
}
/**
* 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {
System.out.println("exceptionCaught");
cause.printStackTrace();
ctx.close();//抛出异常，断开与客户端的连接
}
/**
* 客户端与服务端第一次建立连接时 执行
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelActive(ctx);
InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = inSocket.getAddress().getHostAddress();
System.out.println(clientIp);
}
/**
* 客户端与服务端 断连时 执行
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelInactive(ctx);
InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = inSocket.getAddress().getHostAddress();
ctx.close(); //断开连接时，必须关闭，否则造成资源浪费
System.out.println("channelInactive:"+clientIp);
Global.getInstance().canTcpConnected = false;
}
}这里做的修改较多主要是修改channelRead从服务端收到新的数据时的回调方法
会将数据进行按照16进制读取和解析为字符串，并作为对接的原始数据进行存储。
还会将数据按照16进制解析并获取对应位的字符并赋值到对象保存到数据库。
其中用到的相关工具类方法，又封装了一个NettyConnectHelper
其中包含用来发起连接以及16进制解析和转换对象的相关方法
package com.badao.demo.netty;
import com.badao.demo.entity.StallVo;
public class NettyConnectHelper {
/**
* 发起连接
*/
public static void doConnect(){
try {
/**
* 使用异步注解方式启动netty客户端服务
*/
new BootNettyClient().connect(8600, "127.0.0.1");
}catch (Exception exception){
System.out.println("tcp连接异常");
}
}
/**
* 接收字节数据并转换为16进制字符串
* @param by
* @return
*/
public static String receiveHexToString(byte[] by) {
try {
String str = bytes2Str(by);
str = str.toUpperCase();
return str;
} catch (Exception ex) {
ex.printStackTrace();
System.out.println("接收字节数据并转为16进制字符串异常");
}
return null;
}
/**
* 字节数组转换为16进制字符串
* @param src
* @return
*/
public static String bytes2Str(byte[] src){
StringBuilder stringBuilder = new StringBuilder("");
if (src == null || src.length <= 0) {
return null;
}
for (int i = 0; i < src.length; i++) {
if(i>0){
stringBuilder.append(" ");
}
int v = src[i] & 0xFF;
String hv = Integer.toHexString(v);
if (hv.length() < 2) {
stringBuilder.append(0);
}
stringBuilder.append(hv);
}
return stringBuilder.toString();
}
/**
* 字节转换为16进制字符
* @param src
* @return
*/
public static String byte2Str(byte src){
StringBuilder stringBuilder = new StringBuilder("");
int v = src & 0xFF;
String hv = Integer.toHexString(v);
if (hv.length() < 2) {
stringBuilder.append(0);
}
stringBuilder.append(hv.toUpperCase());
return stringBuilder.toString();
}
/**
* 接收字节数据并转换为对象
* @param by
* @return
*/
public static StallVo receiveHexToObj(byte[] by) {
try {
StallVo stallVo = bytes2Obj(by);
return stallVo;
} catch (Exception ex) {
ex.printStackTrace();
System.out.println("接收字节数据并转为对象异常");
}
return null;
}
/**
* 对象属性赋值
* @param src
* @return
*/
public static StallVo bytes2Obj(byte[] src){
if (src == null || src.length <= 0) {
return null;
}
//依据约定，第一位为区域编号；第四位为车辆状态；第五位为车辆编号
StallVo stallVo = StallVo.builder()
.areaNumber(byte2Str(src[0]))
.carStatus(byte2Str(src[3]))
.carNumber(byte2Str(src[4]))
.build();
return stallVo;
}
}这里跟业务相关挺多，对象属性映射的都是依据对接时的约定。
这里的对象StallVo
package com.badao.demo.entity;
import lombok.Builder;
import lombok.Data;
/**
* 失速保护VO
*/
@Data
@Builder
public class StallVo {
//区域编号
private String areaNumber;
//车辆状态
private String carStatus;
//车辆编号
private String carNumber;
}继续上面的读的处理类
if(Global.getInstance().abnormalCarStatusList.contains(stallVo.getCarStatus()))这里是业务需要，根据传输的数据进行判断，指定位的数据是否为需要的类型数据，只对需要的数据进行存储。
下面附全局单例类Global
package com.badao.demo.global;
import com.badao.demo.enums.CarStatus;
import java.util.ArrayList;
import java.util.List;
public class Global {
//标识当前是否已经连接TCP
public boolean canTcpConnected = false;
//过滤tcp数据，需要的数据类型的枚举变量的list
public List<String> abnormalCarStatusList = new ArrayList<String>()
{
{
add(CarStatus.OverSpeed.getCode());
add(CarStatus.EmergStop.getCode());
add(CarStatus.StallProtected.getCode());
}
};
private static final Global _global = new Global();
private Global(){};
public static Global getInstance(){
return _global;
}
}关于单例模式的实现可参考
设计模式-单例模式-饿汉式单例模式、懒汉式单例模式、静态内部类在Java中的使用示例:
设计模式-单例模式-饿汉式单例模式、懒汉式单例模式、静态内部类在Java中的使用示例_霸道流氓气质的博客-CSDN博客
其中Global中保存的list是枚举类的相关字段属性
package com.badao.demo.enums;
/**
* 车辆状态
*
*/
public enum CarStatus
{
NormalCar("00", "没有车辆通过或车辆速度正常"), OverSpeed("01", "车辆超速行驶"),EmergStop("02", "车辆急停"), StallProtected("03", "车辆失速保护");
private final String code;
private final String info;
CarStatus(String code, String info)
{
this.code = code;
this.info = info;
}
public String getCode()
{
return code;
}
public String getInfo()
{
return info;
}
}继续上面在解析数据并判断是需要的类型之后，就是封装到存储数据库的相关实体并插入到mysql。
附BusStallProptection
package com.badao.demo.entity;
import lombok.Builder;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;
@Data
@Builder
public class BusStallProptection implements Serializable {
private Integer id;
private String carNumber;
private String carState;
private String stallScope;
private String rawData;
private Date uploadTime;
}封装完之后实现调用mapper的方法插入到数据库。
7、Netty的I/O数据读写处理类BootNettyChannelInboundHandlerAdapter中注入Mapper的方式
I/O数据读写处理类BootNettyChannelInboundHandlerAdapter添加注解@Component
@ChannelHandler.Sharable
@Component
public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{然后正常注入mapper或service
//1.正常注入[记得主类也需要使用@Component注解]
@Autowired
BusStallProptectionMapper busStallProptectionMapper;然后新增初始化构造方法
//2.初始化构造方法一定要有
public BootNettyChannelInboundHandlerAdapter(){
}然后容器初始化时执行如下
//3.容器初始化的时候进行执行-这里是重点
@PostConstruct
public void init() {
bootNettyChannelInboundHandlerAdapter = this;
bootNettyChannelInboundHandlerAdapter.busStallProptectionMapper = this.busStallProptectionMapper;
}前面要声明static变量
public static BootNettyChannelInboundHandlerAdapter bootNettyChannelInboundHandlerAdapter;然后在使用时就可以
bootNettyChannelInboundHandlerAdapter.busStallProptectionMapper.insert(busStallProptection);
8、修改SpringBoot的启动类，使Netty项目启动后进行TCP连接
package com.badao.demo;
import com.badao.demo.netty.NettyConnectHelper;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@MapperScan("com.badao.demo.mapper")
@SpringBootApplication
public class DemoApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication application = new SpringApplication(DemoApplication.class);
application.run(args);
}
@Override
public void run(String... args) {
//如果需要项目一启动就连接则执行，否则通过定时任务执行
NettyConnectHelper.doConnect();
}
}这里将发起连接的操作封装到工具类方法中，并在方法中添加try-catch，避免连接不上tcp导致无法启动。
9、SpringBoot中进行TCP客户端断线检测与自动重连。
这里需要TCP的客户端在断线之后能自动发起重连，且不需重启SpringBoot，所以这里需要借助定时任务的
实现。
新建Task类并进行定时任务实现
package com.badao.demo.task;
import com.badao.demo.global.Global;
import com.badao.demo.netty.NettyConnectHelper;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
@EnableScheduling
public class CheckTcpConnectTask {
@Scheduled(cron = "0/10 * * * * ? ")
public void checkReconnectTcpServer(){
System.out.println("发起重连检测");
if(!Global.getInstance().canTcpConnected){
//进行连接
System.out.println("执行连接");
NettyConnectHelper.doConnect();
}
}
}这里是10秒检测一次前面定义的全局变量，如果未连接则调用发起连接的方法。
该变量默认为false，在建立连接的回调方法BootNettyClient中连接服务端之后将其赋值
为true。
ChannelFuture future = bootstrap.connect(host, port).sync();
if(future.isSuccess()) {
//是否连接tcp成功
Global.getInstance().canTcpConnected = true;
并在断连的回调BootNettyChannelInboundHandlerAdapter的channelInactive赋值为false
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelInactive(ctx);
InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = inSocket.getAddress().getHostAddress();
ctx.close(); //断开连接时，必须关闭，否则造成资源浪费
System.out.println("channelInactive:"+clientIp);
Global.getInstance().canTcpConnected = false;
}
10、断连检测效果
11、TCP服务端、客户端模拟测试工具
第一个是sokit
http://sqdownd.onlinedown.net/down/sokit-1.3-win32-chs.zip
下载之后解压即用
可模拟TCP服务器、客户端、转发器等，F1打开帮助，如果需要发送16进制数据，需要用方括号
包围。
除此之外还要诸多其他模拟和测试工具，比如客户端工具serial，可自行搜索友善串口调试助手。
我有一个字符串input="maybe(thisis|thatwas)some((nice|ugly)(day|night)|(strange(weather|time)))"Ruby中解析该字符串的最佳方法是什么？我的意思是脚本应该能够像这样构建句子:maybethisissomeuglynightmaybethatwassomenicenightmaybethiswassomestrangetime等等，你明白了......我应该一个字符一个字符地读取字符串并构建一个带有堆栈的状态机来存储括号值以供以后计算，还是有更好的方法？也许为此目的准备了一个开箱即用的库？
很好奇，就使用rubyonrails自动化单元测试而言，你们正在做什么？您是否创建了一个脚本来在cron中运行rake作业并将结果邮寄给您？git中的预提交Hook？只是手动调用？我完全理解测试，但想知道在错误发生之前捕获错误的最佳实践是什么。让我们理所当然地认为测试本身是完美无缺的，并且可以正常工作。下一步是什么以确保他们在正确的时间将可能有害的结果传达给您？ 最佳答案 不确定您到底想听什么，但是有几个级别的自动代码库控制:在处理某项功能时，您可以使用类似autotest的内容获得关于哪些有效，哪些无效的即时反馈。要确保您的提
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
我正在使用ruby1.9解析以下带有MacRoman字符的csv文件#encoding:ISO-8859-1#csv_parse.csvName,main-dialogue"Marceu","Giveittohimóhe,hiswife."我做了以下解析。require'csv'input_string=File.read("../csv_parse.rb").force_encoding("ISO-8859-1").encode("UTF-8")#=>"Name,main-dialogue\r\n\"Marceu\",\"Giveittohim\x97he,hiswife.\"\
我收到这个错误:RuntimeError(自动加载常量Apps时检测到循环依赖当我使用多线程时。下面是我的代码。为什么会这样？我尝试多线程的原因是因为我正在编写一个HTML抓取应用程序。对Nokogiri::HTML(open())的调用是一个同步阻塞调用，需要1秒才能返回，我有100,000多个页面要访问，所以我试图运行多个线程来解决这个问题。有更好的方法吗？classToolsController0)app.website=array.join(',')putsapp.websiteelseapp.website="NONE"endapp.saveapps=Apps.order("
我有一个用户工厂。我希望默认情况下确认用户。但是鉴于unconfirmed特征，我不希望它们被确认。虽然我有一个基于实现细节而不是抽象的工作实现，但我想知道如何正确地做到这一点。factory:userdoafter(:create)do|user,evaluator|#unwantedimplementationdetailshereunlessFactoryGirl.factories[:user].defined_traits.map(&:name).include?(:unconfirmed)user.confirm!endendtrait:unconfirmeddoenden
简而言之错误:NOTE:Gem::SourceIndex#add_specisdeprecated,useSpecification.add_spec.Itwillberemovedonorafter2011-11-01.Gem::SourceIndex#add_speccalledfrom/opt/local/lib/ruby/site_ruby/1.8/rubygems/source_index.rb:91./opt/local/lib/ruby/gems/1.8/gems/rails-2.3.8/lib/rails/gem_dependency.rb:275:in`==':und
我正在尝试使用Curbgem执行以下POST以解析云curl-XPOST\-H"X-Parse-Application-Id:PARSE_APP_ID"\-H"X-Parse-REST-API-Key:PARSE_API_KEY"\-H"Content-Type:image/jpeg"\--data-binary'@myPicture.jpg'\https://api.parse.com/1/files/pic.jpg用这个:curl=Curl::Easy.new("https://api.parse.com/1/files/lion.jpg")curl.multipart_form_
我正在使用Ruby，我正在与一个网络端点通信，该端点在发送消息本身之前需要格式化“header”。header中的第一个字段必须是消息长度，它被定义为网络字节顺序中的2二进制字节消息长度。比如我的消息长度是1024。如何将1024表示为二进制双字节？ 最佳答案 Ruby(以及Perl和Python等)中字节整理的标准工具是pack和unpack。ruby的packisinArray.您的长度应该是两个字节长，并且按网络字节顺序排列，这听起来像是n格式说明符的工作:n|Integer|16-bitunsigned,network(bi
华为OD机试题本篇题目：明明的随机数题目输入描述输出描述：示例1输入输出说明代码编写思路最近更新的博客华为od2023|什么是华为od，od薪资待遇，od机试题清单华为OD机试真题大全，用Python解华为机试题|机试宝典【华为OD机试】全流程解析+经验分享,题型分享,防作弊指南华为o