springboot集成socket服务

1，先在配置文件中配置Socket监听的端口，由于JDK1.8中已经整合了Socket服务到java.net包中，因此不需要引入其他依赖了

在application.yml中配置下

#Socket配置
socket:
  port: 8503

2，配置Socket连接类，这一步类似配置一个控制器（Controller），主要用于接收客户端的连接请求，我这里将它写在了控制器类中

package com.linzhuo.app.controller;import com.linzhuo.app.service.IBusinessSocketService;import org.springframework.beans.factory.annotation.Value;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;import javax.annotation.Resource;import java.io.BufferedReader;import java.io.InputStreamReader;import java.io.PrintWriter;import java.net.ServerSocket;import java.net.Socket;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;

/**
 * @author: jss
 * Date: 2023/7/25 上午10:58
 */
@Slf4j
@Component
public class SocketServers {

    //注入被开放的端口
    @Value("${socket.port}")
    private Integer port;

    //这个是业务处理的接口
    @Resource
    private IBusinessSocketService socketService;

    @PostConstruct
    public void socketStart(){
        //直接另起一个线程挂起Socket服务
        new Thread(this::socketServer).start();}

    private void socketServer(){
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        try (ServerSocket serverSocket = new ServerSocket(port)){
            log.info("socket端口在:{}中开启并持续监听=====>", port);while(Boolean.TRUE){
                Socket clientSocket = serverSocket.accept();
                //设置流读取的超时时间,这里设置为10s。超时后自动断开连接
                clientSocket.setSoTimeout(10000);
                //是否与客户端保持持续连接,这行代码在本示例中,并没有作用,因为后面的逻辑处理完成后,会自动断开连接.
                clientSocket.setKeepAlive(Boolean.TRUE);
                log.info("发现客户端连接Socket:{}:{}===========>", clientSocket.getInetAddress().getHostAddress(),
                        clientSocket.getPort());
                //这里通过线程池启动ClientHandler方法中的run方法.
                executorService.execute(new ClientHandler(clientSocket, socketService));}} catch (Exception e){
            log.error("Socket服务启动异常!", e);}}

    public static void main(String[] args) throws Exception {
        requestInfoToSocketServer();}

    private static void requestInfoToSocketServer(){
        try (Socket socket = new Socket("127.0.0.1", 8503);
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true)){
//            Object phone ="15210161743";
            out.write("15210161743");
            out.flush();
            //记得调用shutdownOutput()方法,否则服务端的流读取会一直等待
            socket.shutdownOutput();
            //开始接收服务端的消息
            BufferedReader in= new BufferedReader(new InputStreamReader(socket.getInputStream()));
            System.out.println("接收到服务端的回复:" + in.readLine());} catch (Exception e){
            System.out.println("Socket传输数据异常!" + e.getMessage());}}}

3，配置自定义客户端类，这里也将写在了控制器类中，也可以自己去写在其他文件夹里

package com.linzhuo.app.controller;import com.linzhuo.app.service.IBusinessSocketService;import com.linzhuo.app.util.SnowFlakeUtil;import com.linzhuo.app.util.StringUtil;import lombok.SneakyThrows;import lombok.extern.slf4j.Slf4j;import org.springframework.util.StringUtils;import java.io.IOException;import java.io.PrintWriter;import java.net.Socket;import java.util.concurrent.TimeUnit;

/**
 * @author: jss
 * Date: 2023/7/25 上午11:06
 */
@Slf4j
public class ClientHandler implements Runnable {

    private final Socket clientSocket;

    private final IBusinessSocketService socketService;

    public ClientHandler(Socket clientSocket, IBusinessSocketService socketService){
        this.clientSocket = clientSocket;
        this.socketService = socketService;}

    @Override
    @SneakyThrows
    public void run(){
        //SnowFlakeUtil 雪花ID生成工具类,后面会统一给出
        String logId = SnowFlakeUtil.getId();
        String hostIp = clientSocket.getInetAddress().getHostAddress();
        String port = String.valueOf(clientSocket.getPort());
        try (PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), Boolean.TRUE)){
            //这里的StringUtil是自己写的工具类,后面会统一给出
            String requestInfo = StringUtil.readToStreamToString(clientSocket.getInputStream());if(requestInfo != null){
                log.info("监听到客户端消息:{},监听日志ID为:{}", requestInfo, logId);
                socketService.executeBusinessCode(requestInfo, logId, out);
                clientSocket.shutdownOutput();
                TimeUnit.SECONDS.sleep(1L);}} catch (IOException e){
            log.error("与客户端:[{}:{}]通信异常!错误信息:{}", hostIp, port, e.getMessage());} finally {
            log.info("客户端:[{}:{}]Socket连接已关闭,日志ID为:{}========>", hostIp, port, logId);}}}

4，定义业务接口，service类

package com.linzhuo.app.service;import java.io.PrintWriter;

/**
 * @author: jss
 * Date: 2023/7/25 上午11:11
 */
public interface IBusinessSocketService {
    /**
     * 从Socket中接受消息并处理
     *
     * @param requestInfo 请求报文
     * @param logId       日志ID
     * @param writer      回写给客户端消息的回写类(Socket自带)
     */
    void executeBusinessCode(String requestInfo, String logId, PrintWriter writer);}

5，定义业务实现类，serviceImpl

package com.linzhuo.app.service.impl;import com.linzhuo.app.service.IBusinessSocketService;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Service;import org.springframework.util.StringUtils;import javax.annotation.Resource;import java.io.PrintWriter;

/**
 * @author: jss
 * Date: 2023/7/25 上午11:12
 */
@Slf4j
@Service
public class IBusinessSocketServiceImpl implements IBusinessSocketService {

    @Resource
    LzLoginServiceImpl loginService;
    @Override
    public void executeBusinessCode(String requestInfo, String logId, PrintWriter writer){
        Object responseMsg;
        boolean isSuccess = Boolean.TRUE;
        try {if(StringUtils.isEmpty(requestInfo)){return;}
            //执行你的业务操作
            log.info("接收到的手机为:{}", requestInfo);
//            responseMsg ="回填给客户端的信息,可以是任何格式的对象34";
//这里调用了自己的其他接口，接收客户端发送的手机号 ，进行发送验证码
            Object phoneCode = loginService.getPhoneCode(requestInfo, "1");
            responseMsg = phoneCode;} catch (Exception e){
            isSuccess = Boolean.FALSE;
            e.printStackTrace();
            responseMsg ="回填给客户端的信息(业务处理错误的情况下)";}
        try {
            //将响应报文通过PrintWriter回写给客户端
            writer.println(responseMsg);} catch (Exception e){
            log.error("Socket客户端数据返回异常!当前日志ID:[{}],异常信息:{}", logId, e);}}}

6，两个相关的工具类

第一个SnowFlakeUtil

package com.linzhuo.app.util;

/**************************************************
 * copyright (c)2022
 * created by jss
 * date:       2022-9-18
 * description: 雪花算法工具类
 *
 **************************************************/
public class SnowFlakeUtil {

    private long workerId;
    private long datacenterId;
    private long sequence = 0L;
    private long twepoch = 1288834974657L;                              //  Thu, 04 Nov 2010 01:42:54 GMT 标记时间 用来计算偏移量，距离当前时间不同，得到的数据的位数也不同
    private long workerIdBits = 5L;                                     //  物理节点ID长度
    private long datacenterIdBits = 5L;                                 //  数据中心ID长度
    private long maxWorkerId =-1L ^ (-1L << workerIdBits);             //  最大支持机器节点数0~31，一共32个
    private long maxDatacenterId =-1L ^ (-1L << datacenterIdBits);     //  最大支持数据中心节点数0~31，一共32个
    private long sequenceBits = 12L;                                    //  序列号12位， 4095，同毫秒内生成不同id的最大个数
    private long workerIdShift = sequenceBits;                          //  机器节点左移12位
    private long datacenterIdShift = sequenceBits + workerIdBits;       //  数据中心节点左移17位
    private long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits; //  时间毫秒数左移22位
    private long sequenceMask =-1L ^ (-1L << sequenceBits);                          // 用于和当前时间戳做比较，以获取最新时间
    private long lastTimestamp = -1L;

    //成员类，SnowFlakeUtil的实例对象的保存域
    private static class IdGenHolder {
        private static final SnowFlakeUtil instance = new SnowFlakeUtil();}

    //外部调用获取SnowFlakeUtil的实例对象，确保不可变
    public static SnowFlakeUtil get(){return IdGenHolder.instance;}

    //初始化构造，无参构造有参函数，默认节点都是0
    public SnowFlakeUtil(){
        this(0L, 0L);}

    //设置机器节点和数据中心节点数，都是 0-31
    public SnowFlakeUtil(long workerId, long datacenterId){if(workerId > maxWorkerId || workerId <0){
            throw new IllegalArgumentException(String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));}if(datacenterId > maxDatacenterId || datacenterId <0){
            throw new IllegalArgumentException(String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));}
        this.workerId = workerId;
        this.datacenterId = datacenterId;}

    //线程安全的id生成方法
    @SuppressWarnings("all")
    public synchronized long nextId(){
        //获取当前毫秒数
        long timestamp = timeGen();
        //如果服务器时间有问题(时钟后退) 报错。
        if(timestamp < lastTimestamp){
            throw new RuntimeException(String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));}
        //如果上次生成时间和当前时间相同,在同一毫秒内
        if(lastTimestamp == timestamp){
            //sequence自增，因为sequence只有12bit，所以和sequenceMask相与一下，去掉高位
            sequence =(sequence + 1)& sequenceMask;
            //判断是否溢出,也就是每毫秒内超过4095，当为4096时，与sequenceMask相与，sequence就等于0
            if(sequence ==0){
                //自旋等待到下一毫秒
                timestamp = tilNextMillis(lastTimestamp);}}else{
            //如果和上次生成时间不同,重置sequence，就是下一毫秒开始，sequence计数重新从0开始累加，每个毫秒时间内，都是从0开始计数，最大4095
            sequence = 0L;}
        lastTimestamp = timestamp;
        // 最后按照规则拼出ID 64位
        // 000000000000000000000000000000000000000000  00000            00000       000000000000
        //1位固定整数   time                                       datacenterId   workerId    sequence
        return((timestamp - twepoch)<< timestampLeftShift)|(datacenterId << datacenterIdShift)|(workerId << workerIdShift)| sequence;}

    //比较当前时间和过去时间，防止时钟回退（机器问题），保证给的都是最新时间/最大时间
    protected long tilNextMillis(long lastTimestamp){
        long timestamp = timeGen();while(timestamp <= lastTimestamp){
            timestamp = timeGen();}return timestamp;}

    //获取当前的时间戳（毫秒）
    protected long timeGen(){return System.currentTimeMillis();}

    /**
     * 获取全局唯一编码
     */
    public static String getId(){
        long id= SnowFlakeUtil.get().nextId();return Long.toString(id);}}

第二个工具类为StringUtil

package com.linzhuo.app.util;import lombok.SneakyThrows;import java.io.ByteArrayOutputStream;import java.io.InputStream;

/**
 * <Description> ******
 *
 * @author jss
 * @version 1.0
 * @date 2022/04/17
 */
public class StringUtil {

    @SneakyThrows
    public static String readToStreamToString(InputStream is){
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        byte[] buffer = new byte[2048];
        int bytesRead;
        /*
        这一步可能会卡住,因为客户端如果传输完数据以后,
        如果没有调用socket.shutdownOutput()方法,会导致服务端不知道流是否已传输完毕,
        等待我们之前设置的10S流读取时间后,连接就会被自动关掉
        如果出现这种情况,服务端可以通过其它方式判断。例如换行符或者特殊字符等,只需要在
        while条件中加一个&&判断即可.例如我这里的业务结束标记是字符: ">",那么判断逻辑如下
        若客户端调用了shutdownOutput()方法,则不需要这个判断
        */
        while(!bos.toString().contains(">")&&(bytesRead = is.read(buffer))!= -1){
            bos.write(buffer, 0, bytesRead);}return bos.toString();}}

7,最后测试一下，用Main方法或者另外一个Springboot项目

public static void main(String[] args) throws Exception {
        requestInfoToSocketServer();}

    private static void requestInfoToSocketServer(){
        try (Socket socket = new Socket("127.0.0.1", 8503);
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true)){
//            Object phone ="15210161743";
            out.write("15210161743");
            out.flush();
            //记得调用shutdownOutput()方法,否则服务端的流读取会一直等待
            socket.shutdownOutput();
            //开始接收服务端的消息
            BufferedReader in= new BufferedReader(new InputStreamReader(socket.getInputStream()));
            System.out.println("接收到服务端的回复:" + in.readLine());} catch (Exception e){
            System.out.println("Socket传输数据异常!" + e.getMessage());}}

out.flush();这里需要手动刷新下，才会发送到服务器端，自动刷新不生效 需要手动刷新
自己测试了可以使用