当前位置：
SpringBoot整合Socket _
故人归 | 2025-07-13
129人浏览
前言
前段时间公司一个物联网项目需要通过TCP连接设备收发消息，现在我把代码整理出来，分享一下。
源代码已发布在GitHub
创建Socket
使用ServerSocket绑定IP和端口，
TcpSocket实现Java的Runnable的类，在run方法中使用Accept监听端口是否有客户端发送连接请求，如果有连接来了就创建SocketReceive对象然后将他扔给线程池执行。 |
package com.example.socket_demo.socket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/* */
@Component
@Slf4j
public class TcpSocket implements Runnable {
public Integer port;
private ServerSocket server;
private ExecutorService threadPool;
public TcpSocket() {
try {
port = 8081;
threadPool = Executors.newCachedThreadPool();
server = new ServerSocket(port);
} catch (Exception e) {
log.error(e.getMessage());
}
}
@Override
public void run() {
while (true) {
try {
Socket socket = server.accept();
if (socket != null) {
SocketReceive socketReceive = new SocketReceive(socket);
threadPool.submit(socketReceive);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
__JAVA
---|---
在线连接
AllClientsMap类存放了所有的在线连接，通过hostAddress为key，Socket为Value。
|
package com.example.socket_demo.socket;
import lombok.extern.slf4j.Slf4j;
import java.net.Socket;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@Slf4j
public class AllClientsMap {
/* */
private static final ConcurrentMap<String, Socket> ALLCLIENTS = new ConcurrentHashMap<>();
/* */
public static ConcurrentMap<String, Socket> getAllClients() {
return ALLCLIENTS;
}
/* */
public static Socket getSocketByKey(String key) {
return ALLCLIENTS.get(key);
}
/* */
public static void put(String key, Socket socket) {
ALLCLIENTS.put(key, socket);
log.info("设备Key:{}========ip:{}已加入列表", key, socket.getInetAddress().getHostAddress());
}
/* */
public static void remove(String key) {
ALLCLIENTS.remove(key);
log.info("已移除设备Key:{}", key);
}
/* */
public static int size() {
log.info("当前设备数：{}", ALLCLIENTS.size());
return ALLCLIENTS.size();
}
/* */
public static void print() {
log.info("当前设备列表信息:长度：{}", ALLCLIENTS.size());
ALLCLIENTS.forEach((key, socket) -> {
log.info("设备Key:{}========ip:{}", key, socket.getInetAddress().getHostAddress());
});
}
/* */
public static boolean contains(String key) {
return ALLCLIENTS.containsKey(key);
}
}
__JAVA
---|---
创建SocketReceive
在SocketReceive类中，我们可以执行相关的接收消息，以及业务操作；在第64行代码的位置，可以通过ApplicationContext获取Spring Bean执行业务代码
|
package com.example.socket_demo.socket;
import lombok.extern.slf4j.Slf4j;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
@Slf4j
public class SocketReceive implements Runnable {
private Socket socket;
public SocketReceive() {
}
public SocketReceive(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
while (true) {
if (null == socket) {
log.info("socket为空");
return;
}
boolean isClosed = socket.isClosed();
String hostAddress = socket.getInetAddress().getHostAddress();
if (isClosed) {
log.info("socket检测到关闭了");
if (AllClientsMap.contains(hostAddress)) {
AllClientsMap.remove(hostAddress);
AllClientsMap.print();
}
return;
}
String hostAddress = socket.getInetAddress().getHostAddress();
try {
//建立客户端信息输入流
DataInputStream in = new DataInputStream(socket.getInputStream());
//定义字节数组读取数据
byte[] bytes = new byte[1024];
int len = in.read(bytes);
if (len == -1) {
return;
}
//定义一个新数组copy，解决读取出来的数据字节不够全是0的问题
byte[] bytes1 = new byte[len];
System.arraycopy(bytes, 0, bytes1, 0, len);
log.info("客户端传的byte字节数组：" + printBytesByStringBuilder(bytes1));
String s = new String(bytes1);
log.info("客户端传的byte字节数组转换成字符串打印:" + s);
//转换hex数据
String data = byteArrayToHex(bytes1);
log.info("接收的16进制数据:" + data);
//如果服务端没有保存该socket
if(!AllClientsMap.contains(hostAddress)){
AllClientsMap.put(hostAddress, socket);
AllClientsMap.print();
}
log.debug("客户端" + hostAddress + "发送数据：{}", data);
//执行业务
System.out.println("执行业务");
//从map中获取客户端发送消息
response(AllClientsMap.getSocketByKey(hostAddress), data);
} catch (IOException e) {
if (AllClientsMap.contains(hostAddress)) {
AllClientsMap.remove(hostAddress);
AllClientsMap.print();
}
try {
socket.close();
log.error("{}断开连接", hostAddress);
return;
} catch (IOException ioException) {
log.error(ioException.getMessage());
}
}
}
}
/* */
public static String printBytesByStringBuilder(byte[] bytes) {
StringBuilder stringBuilder = new StringBuilder();
for (byte aByte : bytes) {
stringBuilder.append(byte2String(aByte));
}
return stringBuilder.toString();
}
public static String byte2String(byte b) {
return String.format("%02x ", b);
}
/* */
public static void response(Socket socket, String msg) {
log.debug("向设备IP：{}发送消息：{}", socket.getInetAddress().getHostAddress(), msg);
OutputStream outputStream = null;
try {
outputStream = socket.getOutputStream();
outputStream.write(hexStringToByteArray(msg));
} catch (IOException e) {
try {
socket.close();
} catch (IOException ioException) {
log.error(ioException.getMessage());
}
log.error(e.getMessage());
}
}
/* */
public static String byteArrayToHex(byte[] bytes) {
StringBuilder result = new StringBuilder();
for (int index = 0, len = bytes.length; index <= len - 1; index += 1) {
int char1 = ((bytes[index] >> 4) & 0xF);
char chara1 = Character.forDigit(char1, 16);
int char2 = ((bytes[index]) & 0xF);
char chara2 = Character.forDigit(char2, 16);
result.append(chara1);
result.append(chara2);
}
return result.toString();
}
/* */
public static byte[] hexStringToByteArray(String hexString) {
hexString = hexString.replaceAll(" ", "");
int len = hexString.length();
byte[] bytes = new byte[len / 2];
for (int i = 0; i < len; i += 2) {
// 两位一组，表示一个字节,把这样表示的16进制字符串，还原成一个字节
bytes[i / 2] = (byte) ((Character.digit(hexString.charAt(i), 16) << 4) + Character
.digit(hexString.charAt(i + 1), 16));
}
return bytes;
}
}
__JAVA
---|---
启动
通过继承Spring的InitializingBean类，重写afterPropertiesSet方法，这个方法将在所有的属性被初始化后调用。
然后会创建一个线程执行ServerSocket的监听，初始化我们的TcpSocket对象，一旦Server接收到了连接请求后，会创建一个SocketReceive对象将其扔给线程池执行，在线程池中的SocketReceive对象可以通过ApplicationContext获取Spring Bean执行业务代码。
|
package com.example.socket_demo.socket;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class SpringFinishedListener implements InitializingBean {
@Autowired
private TcpSocket tcpsocket;
@Override
public void afterPropertiesSet() {
Thread serverThread = new Thread(tcpsocket);
serverThread.start();
}
}
__JAVA
---|---
测试工具
这里推荐一个测试工具，还挺好用的。下载连接