Springboot之整合Socket连接案例

2023-07-06 03:49:03 88

Socket连接与硬件通信

一、如何让socket随着springboot项目一起启动

SpringBoot中CommandLineRunner的作用：平常开发中有可能需要实现在项目启动后执行的功能，SpringBoot提供的一种简单的实现方案就是添加一个model并实现CommandLineRunner接口，实现功能的代码放在实现的run方法中

具体实现

importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.boot.CommandLineRunner;
importorg.springframework.stereotype.Component;
importjava.net.ServerSocket;
importjava.net.Socket;
importjava.util.concurrent.ArrayBlockingQueue;
importjava.util.concurrent.ThreadPoolExecutor;
importjava.util.concurrent.TimeUnit;
/**
*@author易水●墨龙吟
*@Description
*@create2019-04-1423:40
*/
@Component
publicclassTestRunnerimplementsCommandLineRunner{
@Autowired
privateSocketPropertiesproperties;
@Override
publicvoidrun(String...args)throwsException{
ServerSocketserver=null;
Socketsocket=null;
server=newServerSocket(properties.getPort());
System.out.println("设备服务器已经开启,监听端口:"+properties.getPort());
ThreadPoolExecutorpool=newThreadPoolExecutor(
properties.getPoolCore(),
properties.getPoolMax(),
properties.getPoolKeep(),
TimeUnit.SECONDS,
newArrayBlockingQueue(properties.getPoolQueueInit()),
newThreadPoolExecutor.DiscardOldestPolicy()
);
while(true){
socket=server.accept();
pool.execute(newServerConfig(socket));
}
}
}

此处使用了自定义的线程池，提高对于socket的客户端处理能力。

二、自定义配置并使用

此处将socket的端口和线程池的一些配置放到application.yml中使用，方便使用和修改

#Socket配置
socket:
#监听端口2323
port:2323
#线程池-保持线程数20
pool-keep:20
#线程池-核心线程数10
pool-core:10
#线程池-最大线程数20
pool-max:30
#线程队列容量10
pool-queue-init:10

importlombok.Getter;
importlombok.Setter;
importlombok.ToString;
importorg.springframework.boot.context.properties.ConfigurationProperties;
importorg.springframework.context.annotation.Configuration;
importorg.springframework.context.annotation.PropertySource;
importorg.springframework.stereotype.Component;
/**
*@author易水●墨龙吟
*@Description
*@create2019-04-1822:35
*/
@Setter
@Getter
@ToString
@Component
@Configuration
@PropertySource("classpath:application.yml")
@ConfigurationProperties(prefix="socket")
publicclassSocketProperties{
privateIntegerport;
privateIntegerpoolKeep;
privateIntegerpoolCore;
privateIntegerpoolMax;
privateIntegerpoolQueueInit;
}

三、Socket对于客户端发来的信息的处理和重发机制

当客户端端连接之后发送信息，如果超时未发送，将会关闭，发送数据有异常将会返回给客户端一个error，让客户端在发送一次数据。

importcom.farm.config.socket.resolve.MessageChain;
importcom.farm.service.EnvironmentService;
importcom.farm.service.impl.EnvironmentServiceImpl;
importjava.io.*;
importjava.net.Socket;
importjava.net.SocketException;
importjava.net.SocketTimeoutException;
importjava.util.Map;
/**
*@author易水●墨龙吟
*@Description
*@create2019-04-1423:21
*/
publicclassServerConfigextendsThread{
privateSocketsocket;
publicServerConfig(Socketsocket){
this.socket=socket;
}
	//获取spring容器管理的类，可以获取到sevrice的类
privateEnvironmentServiceservice=SpringUtil.getBean(EnvironmentServiceImpl.class);
privateStringhandle(InputStreaminputStream)throwsIOException,DataFormException{
byte[]bytes=newbyte[1024];
intlen=inputStream.read(bytes);
if(len!=-1){
StringBufferrequest=newStringBuffer();
request.append(newString(bytes,0,len,"UTF-8"));
System.out.println("接受的数据:"+request);
System.out.println("fromclient..."+request+"当前线程"+Thread.currentThread().getName());
Mapmap=MessageChain.out(request.toString());
System.out.println("处理的数据"+map);
Integerres=service.addEnvironment(map);
if(res==1){
return"ok";
}else{
thrownewDataFormException("数据处理异常");
}
}else{
thrownewDataFormException("数据处理异常");
}
}
@Override
publicvoidrun(){
BufferedWriterwriter=null;
try{
//设置连接超时9秒
socket.setSoTimeout(9000);
System.out.println("客户-"+socket.getRemoteSocketAddress()+"->机连接成功");
InputStreaminputStream=socket.getInputStream();
writer=newBufferedWriter(newOutputStreamWriter(socket.getOutputStream()));
Stringresult=null;
try{
result=handle(inputStream);
writer.write(result);
writer.newLine();
writer.flush();
}catch(IOException|DataFormException|IllegalArgumentExceptione){
writer.write("error");
writer.newLine();
writer.flush();
System.out.println("发生异常");
try{
System.out.println("再次接受!");
result=handle(inputStream);
writer.write(result);
writer.newLine();
writer.flush();
}catch(DataFormException|SocketTimeoutExceptionex){
System.out.println("再次接受,发生异常,连接关闭");
}
}
}catch(SocketExceptionsocketException){
socketException.printStackTrace();
try{
writer.close();
}catch(IOExceptionioException){
ioException.printStackTrace();
}
}catch(IOExceptione){
e.printStackTrace();
}finally{
try{
writer.close();
}catch(IOExceptione){
e.printStackTrace();
}
}
}
}

在此处有一个坑，如果客户端是用C/C++编写的，必须使用如下方法：

byte[]bytes=newbyte[1024];
intlen=inputStream.read(bytes);

如果使用readLine或者DataInputStreamdataInputStream=newDataInputStream(socket.getInputStream())这样会出现使用TCP连接助手，客户端发送数据收不到。

四、如何在普通类中使用Spring注入类

这里需要使用一个工具类。

importorg.springframework.beans.BeansException;
importorg.springframework.context.ApplicationContext;
importorg.springframework.context.ApplicationContextAware;
importorg.springframework.stereotype.Component;
/**
*@author易水●墨龙吟
*@Description
*@create2019-04-150:01
*/
@Component
publicclassSpringUtilimplementsApplicationContextAware{
privatestaticApplicationContextapplicationContext;
@Override
publicvoidsetApplicationContext(ApplicationContextapplicationContext)throwsBeansException{
if(SpringUtil.applicationContext==null){
SpringUtil.applicationContext=applicationContext;
}
}
/**
*获取applicationContext
*@return
*/
publicstaticApplicationContextgetApplicationContext(){
returnapplicationContext;
}
/**
*通过name获取Bean.
*@paramname
*@return
*/
publicstaticObjectgetBean(Stringname){
returngetApplicationContext().getBean(name);
}
/**
*通过class获取Bean.
*@paramclazz
*@param
*@return
*/
publicstaticTgetBean(Classclazz){
returngetApplicationContext().getBean(clazz);
}
/**
*通过name,以及Clazz返回指定的Bean
*@paramname
*@paramclazz
*@param
*@return
*/
publicstaticTgetBean(Stringname,Classclazz){
returngetApplicationContext().getBean(name,clazz);
}
}

补充：springboot下websocket前台后端数据长连接

首先导入依赖


org.springframework.boot
spring-boot-starter-websocket


org.springframework.security
spring-security-messaging

spring-security-messaging是后面继承AbstractSecurityWebSocketMessageBrokerConfigurer需要用到的依赖

WebSocketConfig

@Configuration
@EnableWebSocketMessageBroker//此注解表示使用STOMP协议来传输基于消息代理的消息，此时可以在@Controller类中使用@MessageMapping
publicclassWebSocketConfigimplementsWebSocketMessageBrokerConfigurer{
@Override
publicvoidregisterStompEndpoints(StompEndpointRegistryregistry){
/**
*注册Stomp的端点
*addEndpoint：添加STOMP协议的端点。这个HTTPURL是供WebSocket或SockJS客户端访问的地址
*withSockJS：指定端点使用SockJS协议
*/
registry.addEndpoint("/websocket/tracker")//物流消息通道,
.setAllowedOrigins("*")//允许跨域,里面路径可以设定
.withSockJS()//指定协议
.setInterceptors(httpSessionHandshakeInterceptor());//设置拦截器()
}
@Override
publicvoidconfigureMessageBroker(MessageBrokerRegistryregistry){
/**
*配置消息代理
*启动简单Broker，消息的发送的地址符合配置的前缀来的消息才发送到这个broker
*/
registry.enableSimpleBroker("/topic","/user");
}
//拦截器
@Bean
publicHandshakeInterceptorhttpSessionHandshakeInterceptor(){
returnnewHandshakeInterceptor(){
@Override
publicbooleanbeforeHandshake(ServerHttpRequestrequest,ServerHttpResponseresponse,WebSocketHandlerwsHandler,Mapattributes)throwsException{
//可以在这里先判断登录是否合法
returntrue;
}
@Override
publicvoidafterHandshake(ServerHttpRequestrequest,ServerHttpResponseresponse,WebSocketHandlerwsHandler,Exceptionexception){
//握手成功后,
}
};
}
}

WebsocketSecurityConfiguration

@Configuration
publicclassWebsocketSecurityConfigurationextendsAbstractSecurityWebSocketMessageBrokerConfigurer{
@Override
protectedvoidconfigureInbound(MessageSecurityMetadataSourceRegistrymessages){
messages
.nullDestMatcher().authenticated()
.simpDestMatchers("/topic/**").authenticated()
.simpDestMatchers("/user/**").authenticated()
.simpTypeMatchers(SimpMessageType.MESSAGE,SimpMessageType.SUBSCRIBE).denyAll()
//catchall
.anyMessage().denyAll();
}
/**
*DisablesCSRFforWebsockets.
*/
@Override
protectedbooleansameOriginDisabled(){
returntrue;
}
}

WebSocketResource

packagecom.gleam.shopmall.web.rest;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.context.ApplicationListener;
importorg.springframework.messaging.handler.annotation.MessageMapping;
importorg.springframework.messaging.handler.annotation.SendTo;
importorg.springframework.messaging.simp.SimpMessageHeaderAccessor;
importorg.springframework.messaging.simp.SimpMessageMappingInfo;
importorg.springframework.messaging.simp.SimpMessageSendingOperations;
importorg.springframework.stereotype.Controller;
importorg.springframework.web.socket.messaging.SessionDisconnectEvent;
@Controller
publicclassWebSocketResource{
privatestaticfinalLoggerlog=LoggerFactory.getLogger(WebSocketResource.class);
@Autowired
SimpMessageSendingOperationsmessagingTemplate;
//此方法适用于网页聊天室,从前端接收数据,返回订阅者(前端)
@MessageMapping("/welcome")//指定要接收消息的地址，类似@RequestMapping
@SendTo("/topic/getResponse")//默认消息将被发送到与传入消息相同的目的地，但是目的地前面附加前缀（默认情况下为“/topic”}
publicStringsay(Stringmessage)throwsException{
returnmessage;
}
//发送指定用户(直接从后端发送数据到前端)
publicvoidsendToUser(Stringlogin,Stringchannel,Stringinfo){
log.debug("[ToUser]WEBSOCKET发送消息,username={},info={}",login,info);
this.messagingTemplate.convertAndSendToUser(login,channel,info);
log.debug("[ToUser]WEBSOCKET发送消息：完成");
}
//发送所有订阅的(直接从后端发送数据到前端)
publicvoidsend(Stringchannel,Stringinfo){
log.debug("[ToAll]WEBSOCKET发送消息,info={}",info);
//this.messagingTemplate.convertAndSend(channel,info);
this.messagingTemplate.convertAndSend("/topic/getResponse","接收到了吗?");
log.debug("[ToAll]WEBSOCKET发送消息：完成");
}
}

前端html









SpringBoot+WebSocket+广播式

varstompClient=null;
functionsetConnected(connected){
document.getElementById('connect').disabled=connected;
document.getElementById('disconnect').disabled=!connected;
document.getElementById('conversationDiv').style.visibility=connected?'visible':'hidden';
$('#response').html();
}
functionconnect(){
//websocket的连接地址，此值等于WebSocketConfig中registry.addEndpoint("/websocket/tracker").withSockJS()配置的地址,
//这里如果是微服务或者远端,需要全路径
varsocket=newSockJS('/websocket/tracker');//1
stompClient=Stomp.over(socket);//2
stompClient.connect({},function(frame){//3
setConnected(true);
console.log('开始进行连接Connected:'+frame);
//客户端订阅消息的目的地址：此值等于WebSocketResource中@SendTo("/topic/getResponse")注解的里配置的值
stompClient.subscribe('/topic/getResponse',function(respnose){//4
showResponse(respnose.body);
});
});
}
functiondisconnect(){
if(stompClient!=null){
stompClient.disconnect();
}
setConnected(false);
console.log("Disconnected");
}
functionsendName(){
varname=$('#name').val();
stompClient.send("/welcome",{},returnCitySN['cip']+":"+name);//JSON.stringify(name)
}
functionshowResponse(message){
varresponse=$("#response");
response.html(message+"
"+response.html());
}






连接
断开连接


输入内容
发送

```

以上为个人经验，希望能给大家一个参考，也希望大家多多支持毛票票。如有错误或未考虑完全的地方，望不吝赐教。

Springboot之整合Socket连接案例

一、如何让socket随着springboot项目一起启动

二、自定义配置并使用

三、Socket对于客户端发来的信息的处理和重发机制

四、如何在普通类中使用Spring注入类

热门推荐

随机推荐