Spring Boot集成netty實(shí)現(xiàn)客戶端服務(wù)端交互示例詳解

更新時(shí)間：2018年12月27日 08:38:47 作者：琴_酒

這篇文章主要給大家介紹了關(guān)于Spring Boot集成netty實(shí)現(xiàn)客戶端服務(wù)端交互的相關(guān)資料，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

前言

Netty 是一個(gè)高性能的 NIO 網(wǎng)絡(luò)框架，本文主要給大家介紹了關(guān)于SpringBoot集成netty實(shí)現(xiàn)客戶端服務(wù)端交互的相關(guān)內(nèi)容，下面來(lái)一起看看詳細(xì)的介紹吧

看了好幾天的netty實(shí)戰(zhàn),慢慢摸索,雖然還沒(méi)有摸著很多門道,但今天還是把之前想加入到項(xiàng)目里的

一些想法實(shí)現(xiàn)了,算是有點(diǎn)信心了吧(講真netty對(duì)初學(xué)者還真的不是很友好......)

首先,當(dāng)然是在SpringBoot項(xiàng)目里添加netty的依賴了,注意不要用netty5的依賴,因?yàn)橐呀?jīng)廢棄了

<!--netty-->
<dependency>
 <groupId>io.netty</groupId>
 <artifactId>netty-all</artifactId>
 <version>4.1.32.Final</version>
</dependency>

將端口和IP寫入application.yml文件里,我這里是我云服務(wù)器的內(nèi)網(wǎng)IP,如果是本機(jī)測(cè)試,用127.0.0.1就ok

netty:
 port: 7000
 url: 172.16.0.7

在這之后,開(kāi)始寫netty的服務(wù)器,這里服務(wù)端的邏輯就是將客戶端發(fā)來(lái)的信息返回回去

因?yàn)椴捎靡蕾囎⑷氲姆椒▽?shí)例化netty,所以加上@Component注釋

package com.safelocate.app.nettyServer;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

@Component
public class NettyServer {
 //logger
 private static final Logger logger = Logger.getLogger(NettyServer.class);
 public void start(InetSocketAddress address){
 EventLoopGroup bossGroup = new NioEventLoopGroup(1);
 EventLoopGroup workerGroup = new NioEventLoopGroup();
 try {
  ServerBootstrap bootstrap = new ServerBootstrap()
   .group(bossGroup,workerGroup)
   .channel(NioServerSocketChannel.class)
   .localAddress(address)
   .childHandler(new ServerChannelInitializer())
   .option(ChannelOption.SO_BACKLOG, 128)
   .childOption(ChannelOption.SO_KEEPALIVE, true);
  // 綁定端口，開(kāi)始接收進(jìn)來(lái)的連接
  ChannelFuture future = bootstrap.bind(address).sync();
  logger.info("Server start listen at " + address.getPort());
  future.channel().closeFuture().sync();
 } catch (Exception e) {
  e.printStackTrace();
  bossGroup.shutdownGracefully();
  workerGroup.shutdownGracefully();
 }
 }

}

當(dāng)然,這里的ServerChannelInitializer是我自己定義的類,這個(gè)類是繼承ChannelInitializer<SocketChannel>的,里面設(shè)置出站和入站的編碼器和解碼器

package com.safelocate.app.nettyServer;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
 @Override
 protected void initChannel(SocketChannel channel) throws Exception {
 channel.pipeline().addLast("decoder",new StringDecoder(CharsetUtil.UTF_8));
 channel.pipeline().addLast("encoder",new StringEncoder(CharsetUtil.UTF_8));
 channel.pipeline().addLast(new ServerHandler());
 }
}

最好注意被別decoder和encoder寫成了一樣的,不然會(huì)出問(wèn)題(我之前就是不小心都寫成了StringDecoder...)

在這之后就是設(shè)置ServerHandler來(lái)處理一些簡(jiǎn)單的邏輯了

package com.safelocate.app.nettyServer;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;

public class ServerHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelActive(ChannelHandlerContext ctx) {
 System.out.println("channelActive----->");
 }


 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("server channelRead......");
 System.out.println(ctx.channel().remoteAddress()+"----->Server :"+ msg.toString());
 //將客戶端的信息直接返回寫入ctx
 ctx.write("server say :"+msg);
 //刷新緩存區(qū)
 ctx.flush();
 }

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 cause.printStackTrace();
 ctx.close();
 }
}

準(zhǔn)備工作到這里,現(xiàn)在要做到就是去啟動(dòng)這個(gè)程序

將AppApplication實(shí)現(xiàn)CommandLineRunner這個(gè)接口,這個(gè)接口可以用來(lái)再啟動(dòng)SpringBoot時(shí)同時(shí)啟動(dòng)其他功能,比如配置,數(shù)據(jù)庫(kù)連接等等

然后重寫run方法,在run方法里啟動(dòng)netty服務(wù)器,Server類用@AutoWired直接實(shí)例化

package com.safelocate.app;

import com.safelocate.app.nettyServer.NettyServer;
import io.netty.channel.ChannelFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.net.InetAddress;
import java.net.InetSocketAddress;
@SpringBootApplication
public class AppApplication implements CommandLineRunner {

 @Value("${netty.port}")
 private int port;

 @Value("${netty.url}")
 private String url;

 @Autowired
 private NettyServer server;

 public static void main(String[] args) {
 SpringApplication.run(AppApplication.class, args);
 }
 @Override
 public void run(String... args) throws Exception {
 InetSocketAddress address = new InetSocketAddress(url,port);
 System.out.println("run .... . ... "+url);
 server.start(address);
 }
}

ok,到這里服務(wù)端已經(jīng)寫完,本地我也已經(jīng)測(cè)試完,現(xiàn)在需要打包部署服務(wù)器,當(dāng)然這個(gè)程序只為練手...

控制臺(tái)輸入mvn clean package -D skipTests 然后將jar包上傳服務(wù)器,在這之后,需要在騰訊云/阿里云那邊配置好安全組,將之前yml文件里設(shè)定的端口的入站

規(guī)則設(shè)置好,不然訪問(wèn)會(huì)被拒絕

之后java -jar命令運(yùn)行,如果需保持后臺(tái)一直運(yùn)行就用nohup命令,可以看到程序已經(jīng)跑起來(lái)了,等待客戶端連接交互

之后就是寫客戶端了,客戶端其實(shí)是依葫蘆畫(huà)瓢,跟上面類似

Handler

package client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class ClientHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelActive(ChannelHandlerContext ctx) {
 System.out.println("ClientHandler Active");
 }

 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) {
 System.out.println("--------");
 System.out.println("ClientHandler read Message:"+msg);
 }


 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
 cause.printStackTrace();
 ctx.close();
 }

}

ChannelInitializer

package client;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
 protected void initChannel(SocketChannel channel) throws Exception {
 ChannelPipeline p = channel.pipeline();
 p.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
 p.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
 p.addLast(new ClientHandler());
 }
}

主函數(shù)所在類,即客戶端

package client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class Client {
 static final String HOST = System.getProperty("host", "服務(wù)器的IP地址");
 static final int PORT = Integer.parseInt(System.getProperty("port", "7000"));
 static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));

 public static void main(String[] args) throws Exception {
 sendMessage("hhhh");
 }
 public static void sendMessage(String content) throws InterruptedException{
 // Configure the client.
 EventLoopGroup group = new NioEventLoopGroup();
 try {
  Bootstrap b = new Bootstrap();
  b.group(group)
   .channel(NioSocketChannel.class)
   .option(ChannelOption.TCP_NODELAY, true)
   .handler(new ChannelInitializer<SocketChannel>() {
   @Override
   public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    p.addLast("decoder", new StringDecoder());
    p.addLast("encoder", new StringEncoder());
    p.addLast(new ClientHandler());
   }
   });

  ChannelFuture future = b.connect(HOST, PORT).sync();
  future.channel().writeAndFlush(content);
  future.channel().closeFuture().sync();
 } finally {
  group.shutdownGracefully();
 }
 }

}

啟動(dòng)客戶端,這里就是簡(jiǎn)單發(fā)送一條"hhhh",可以看到客戶端已經(jīng)收到服務(wù)器發(fā)來(lái)的信息