使用java項(xiàng)目搭建一個(gè)netty服務(wù)

更新時(shí)間：2024年10月30日 11:50:04 作者：傀儡師

這篇文章主要為大家詳細(xì)介紹了如何使用java項(xiàng)目搭建一個(gè)netty服務(wù),文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

映入依賴，只要保證有這個(gè)依賴，就不需要單獨(dú)引入依賴，支持多個(gè)端口直連，支持多個(gè)實(shí)現(xiàn)層解析數(shù)據(jù)，

  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
  <version>3.3.4</version>

yml配置

# TCP設(shè)備對(duì)接
iot:
  device:
    port1: 1883
    port2: 1885

package com.cqcloud.platform.handler;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.cqcloud.platform.service.IotNbIotMqttService;
import com.cqcloud.platform.service.IotPushService;
import com.cqcloud.platform.service.impl.IotNbIotServiceImpl;
import com.cqcloud.platform.service.impl.IotPushServiceImpl;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import jakarta.annotation.PostConstruct;
/**
 * @author weimeilayer@gmail.com ?
 * @date ???? 2022年3月8日???? ????
 */
@Component
public class NettyTcpServer {
	/**
	 * 用于自設(shè)備1協(xié)議端口
	 */
    private static int PORT1;
    /**
     * 來(lái)自設(shè)備2協(xié)議端口
     */
    private static int PORT2;

    @Value("${iot.device.port1}")
    public int port1Value;

    @Value("${iot.device.port2}")
    public int port2Value;
    
    @PostConstruct
    public void init() {
        PORT1 = port1Value;
        PORT2 = port2Value;
    }

	public void start() throws Exception {
		final NioEventLoopGroup bossGroup = new NioEventLoopGroup();
	    final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
	    try {
	        ServerBootstrap bootstrap = new ServerBootstrap();
	        // 創(chuàng)建 MqttService 和 MqttPushService 實(shí)例
	        IotNbIotMqttService iotNbIotMqttService = new IotNbIotServiceImpl();
	        IotPushService iotPushService = new IotPushServiceImpl();

	        bootstrap.group(bossGroup, workerGroup)
	            .channel(NioServerSocketChannel.class)
	            .childHandler(new ChannelInitializer<SocketChannel>() {
	                @Override
	                protected void initChannel(SocketChannel ch) {
	                    ChannelPipeline pipeline = ch.pipeline();
	                    // 直接使用 ByteBuf，無(wú)需編碼器和解碼器
	                    // 根據(jù)端口注入不同的服務(wù)
	                    if (ch.localAddress().getPort() == PORT1) {
	                        pipeline.addLast(new TcpIotNbServerHandler(iotNbIotMqttService)); // 業(yè)務(wù)邏輯處理器
	                    } else if (ch.localAddress().getPort() == PORT2) {
	                        pipeline.addLast(new TcpIotServerHandler(iotPushService)); // 新處理器
	                    }
	                }
	            });

	        // 綁定第一個(gè)端口并啟動(dòng)
	        ChannelFuture future1 = bootstrap.bind(PORT1).sync();
	        // 綁定第二個(gè)端口并啟動(dòng)
	        ChannelFuture future2 = bootstrap.bind(PORT2).sync();

	        // 等待服務(wù)器關(guān)閉
	        future1.channel().closeFuture().sync();
	        future2.channel().closeFuture().sync();
	    } finally {
	        // 優(yōu)雅地關(guān)閉線程池
	        workerGroup.shutdownGracefully();
	        bossGroup.shutdownGracefully();
	    }
	}
}

啟動(dòng)類需要

public static void main(String[] args) throws IOException {
		ConfigurableEnvironment env = new SpringApplication(DynamicYearningApplication.class).run(args).getEnvironment();
		String envPort = env.getProperty("server.port");
		String port = Objects.isNull(envPort) ? "8000" : envPort;
		String envContext = env.getProperty("server.servlet.context-path");
		String contextPath = Objects.isNull(envContext) ? "" : envContext;
		String path = port + contextPath + "/doc.html";
		String externalAPI = InetAddress.getLocalHost().getHostAddress();
		Console.log("Access URLs:\n\t-------------------------------------------------------------------------\n\tLocal-swagger: \t\thttp://127.0.0.1:{}\n\tExternal-swagger: \thttp://{}:{}\n\t-------------------------------------------------------------------------",path, externalAPI, path);
		// 加上以下代碼
		NettyTcpServer server = new NettyTcpServer();
		try {
			server.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

創(chuàng)建TcpIotServerHandler

package com.cqcloud.platform.handler;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.StringPool;
import com.cqcloud.platform.entity.IotCommandRecords;
import com.cqcloud.platform.service.IotPushService;
import com.cqcloud.platform.utils.DeviceActionParser;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * 設(shè)備協(xié)議
 * @author weimeilayer@gmail.com ?
 * @date ???? 2022年3月8日 ???? ????
 */
@Slf4j
public class TcpIotServerHandler extends SimpleChannelInboundHandler<ByteBuf>  {

	// 接口注入
	private final IotPushService iotPushService;

	public TcpIotServerHandler(IotPushService iotPushService) {
		this.iotPushService = iotPushService;
	}
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
		byte[] byteArray;
		if (in.readableBytes() <= 0) {
			in.release();
			return;
		}
		byteArray = new byte[in.readableBytes()];
		in.readBytes(byteArray);
		if (byteArray.length <= 0) {
			in.release();
			return;
		}
		// 將消息傳遞給 iotPushService
		iotPushService.pushMessageArrived(byteArray);
	}
	// 發(fā)送響應(yīng)的統(tǒng)一輔助方法
	private void sendResponse(ChannelHandlerContext ctx, String hexResponse) {
		byte[] responseBytes = hexStringToByteArray(hexResponse);
		ByteBuf responseBuffer = Unpooled.copiedBuffer(responseBytes);
		ctx.writeAndFlush(responseBuffer);
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		// 打印異常堆棧跟蹤，便于調(diào)試和錯(cuò)誤排查
		cause.printStackTrace();
		// 關(guān)閉當(dāng)前的通道，釋放相關(guān)資源
		ctx.close();
	}
}

創(chuàng)建 TcpIotNbServerHandler

package com.cqcloud.platform.handler;

import com.cqcloud.platform.service.IotNbIotMqttService;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * NB-IOT CAT1數(shù)據(jù)格協(xié)議
 * 
 * @author weimeilayer@gmail.com
 * @date ????2022年3月8日????????
 */
public class TcpIotNbServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
	private final IotNbIotMqttService iotNbIotMqttService;

	// 構(gòu)造函數(shù)注入 MqttService
	public TcpIotNbServerHandler(IotNbIotMqttService iotNbIotMqttService) {
		this.iotNbIotMqttService = iotNbIotMqttService;
	}

	@Override
	public void channelRead0(ChannelHandlerContext ctx,ByteBuf in) {
		byte[] byteArray;
		if (in.readableBytes() <= 0) {
			in.release();
			return;
		}
		byteArray = new byte[in.readableBytes()];
		in.readBytes(byteArray);
		if (byteArray.length <= 0) {
			in.release();
			return;
		}
	    // 將 byte[] 數(shù)據(jù)傳遞給 iotNbIotMqttService
	    iotNbIotMqttService.messageArrived(byteArray); 
		//發(fā)送固定事件默認(rèn)回復(fù)
        sendResponse(ctx);
	}
	
	// 發(fā)送響應(yīng)的統(tǒng)一輔助方法
    private void sendResponse(ChannelHandlerContext ctx) {
    	// 回復(fù)客戶端--向設(shè)備回復(fù)AAAA8001（設(shè)備將保持20秒不休眠）,平臺(tái)盡量在10秒
    	byte[] responseBytes = new byte[] { (byte) 0xAA, (byte) 0xAA, (byte) 0x80, (byte) 0x01 };
        ByteBuf responseBuffer = Unpooled.copiedBuffer(responseBytes);
        ctx.writeAndFlush(responseBuffer);
    }
    
    //將響應(yīng)消息轉(zhuǎn)換為字節(jié)數(shù)組
    public static byte[] hexStringToByteArray(String s) {
        int len = s.length();
        byte[] data = new byte[len / 2];
        for (int i = 0; i < len; i += 2) {
            data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4)
                    + Character.digit(s.charAt(i + 1), 16));
        }
        return data;
    }
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		cause.printStackTrace();
		ctx.close();
	}
}

創(chuàng)建接口類IotPushService

package com.cqcloud.platform.service;

/**
 * @author weimeilayer@gmail.com
 * @date ????2022年3月8日????????
 */
public interface IotPushService {
	public void pushMessageArrived(byte[] message);
}

創(chuàng)建IotNbIotMqttService 類

package com.cqcloud.platform.service;

/**
 * @author weimeilayer@gmail.com
 * @date ????2022年3月8日????????
 */
public interface IotNbIotMqttService {

	public void messageArrived(byte[] message);
}

創(chuàng)建實(shí)現(xiàn)類IotNbIotServiceImpl

package com.cqcloud.platform.service.impl;

import org.springframework.stereotype.Service;

import com.cqcloud.platform.service.IotNbIotMqttService;
import com.cqcloud.platform.utils.DataParser;

import lombok.AllArgsConstructor;

/**
 * @author weimeilayer@gmail.com
 * @date ????2022年3月8日????????
 */
@Service
@AllArgsConstructor
public class IotNbIotServiceImpl implements IotNbIotMqttService {

	@Override
	public void messageArrived(byte[] message) {
		// 將 byte 數(shù)組轉(zhuǎn)換為十六進(jìn)制字符串  
		String convertData = printByteArray(message);
		// 打印字節(jié)數(shù)組內(nèi)容
		System.out.println("來(lái)自于xxx數(shù)據(jù)格式協(xié)議的1883端口的數(shù)據(jù)字節(jié)數(shù)組內(nèi)容:"+ convertData);
        //調(diào)用解析方法
        dispatchMessage(convertData);
	}
	
	// 將 byte[] 轉(zhuǎn)換為十六進(jìn)制字符串的輔助方法  
    public static String bytesToHex(byte[] bytes) {  
        StringBuilder hex = new StringBuilder();  
        for (byte b : bytes) {  
            // 將每個(gè)字節(jié)轉(zhuǎn)換為兩位的十六進(jìn)制表示  
            hex.append(String.format("%02X", b));  
        }  
        return hex.toString();  
    }  

	public static String printByteArray(byte[] byteArray) {
		StringBuilder hexString = new StringBuilder();
		for (byte b : byteArray) {
			// 將字節(jié)轉(zhuǎn)換為無(wú)符號(hào)的十六進(jìn)制字符串，去掉空格
			hexString.append(String.format("%02X", b & 0xFF));
		}
		System.out.println("Byte Array (Hex): " + hexString.toString());
		return hexString.toString();
	}
	
	public void dispatchMessage(String byteArray) {
        String prefix = byteArray.substring(0, 2);  
        // 根據(jù) messageID 進(jìn)行判斷
          System.out.println("來(lái)自于數(shù)據(jù)格式協(xié)議來(lái)自于1883端口的數(shù)據(jù)處理消息:" +byteArray);
	}
}

創(chuàng)建 IotPushServiceImpl

package com.cqcloud.platform.service.impl;

import org.springframework.stereotype.Service;

import com.cqcloud.platform.service.IotPushService;
import com.cqcloud.platform.utils.DeviceActionParser;

import lombok.AllArgsConstructor;

/**
 * 發(fā)送指令實(shí)現(xiàn)類
 * @author weimeilayer@gmail.com
 * @date ????2022年3月8日????????
 */
@Service
@AllArgsConstructor
public class IotPushServiceImpl implements IotPushService {

	@Override
	public void pushMessageArrived(byte[] message) {
		// 解析字節(jié)數(shù)組
		System.out.println("來(lái)自物聯(lián)網(wǎng)平臺(tái)的設(shè)備協(xié)議于1885端口的數(shù)據(jù)設(shè)備返回的的內(nèi)容處理");
		//打印數(shù)據(jù)
		printByteArray(message);
		//調(diào)用解析方法
		dispatchMessage(message);
	}

	//設(shè)備回復(fù)的接受內(nèi)容
	public static void dispatchMessage(byte[] byteArray) {
       
	}
	
	public static void printByteArray(byte[] byteArray) {
		StringBuilder hexString = new StringBuilder();
		for (byte b : byteArray) {
			// 將字節(jié)轉(zhuǎn)換為無(wú)符號(hào)的十六進(jìn)制字符串，去掉空格
			hexString.append(String.format("%02X", b & 0xFF));
		}
		System.out.println("Byte Array (Hex): " + hexString.toString());
	}

	 // 將十六進(jìn)制字符串轉(zhuǎn)換為字節(jié)數(shù)組的實(shí)用方法
    public static byte[] stringToBytes(String s) {
        int len = s.length();
        byte[] data = new byte[len / 2];
        for (int i = 0; i < len; i += 2) {
            data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4)
                                  + Character.digit(s.charAt(i+1), 16));
        }
        return data;
    }
	// 提取設(shè)備類型的十六進(jìn)制字符串
    private static String extractDeviceTypeHex(byte[] byteArray) {
        // 轉(zhuǎn)換為十六進(jìn)制字符串
        String hexString = bytesToHex(byteArray);
        // 提取設(shè)備類型
        return hexString.substring(10, 12); // 設(shè)備類型的位數(shù)
    }

    // 輔助方法：將字節(jié)數(shù)組轉(zhuǎn)換為十六進(jìn)制字符串
    private static String bytesToHex(byte[] bytes) {
        StringBuilder hexString = new StringBuilder();
        for (byte b : bytes) {
            String hex = Integer.toHexString(0xFF & b);
            if (hex.length() == 1) {
                hexString.append('0'); // 確保每個(gè)字節(jié)都為兩位
            }
            hexString.append(hex);
        }
        return hexString.toString().toUpperCase(); // 返回大寫格式
    }

    // 將十六進(jìn)制字符串轉(zhuǎn)換為 byte
    private static byte hexStringToByte(String hex) {
        return (byte) Integer.parseInt(hex, 16);
    }
}

然后使用網(wǎng)絡(luò)根據(jù)助手請(qǐng)求。