netty整合spring

痛定思痛。 2022-04-13 06:17 204阅读 0赞

首先注意一个大坑，你的netty的jar是4.xxx不，如果是请注意一定要这个注解

import io.netty.channel.ChannelHandler.Sharable;
    @Sharable
    这个注解的功能是，有了它才能连接多个handler否则只能连接一个handler，我被这个坑了快一个星期了，每次都是第一次连接好的，第二次连接就报错
    java.io.IOException: 你的主机中的软件中止了一个已建立的连接。
    	at sun.nio.ch.SocketDispatcher.read0(Native Method)
    	at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:43)
    	at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
    	at sun.nio.ch.IOUtil.read(IOUtil.java:192)
    	at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
    	at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:288)
    	at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1100)
    	at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:367)
    	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:118)
    	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:610)
    	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:551)
    	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:465)
    	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:437)
    	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:873)
    	at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
    	at java.lang.Thread.run(Thread.java:745)

现在说怎么整合到spring中去

1.实现InitializingBean接口目的是启动的时候运行afterPropertiesSet这个方法

@Component
    public class NettyService implements InitializingBean {
    
    	@Autowired
    	private EchoService echoService;
    	
    	@Override
    	public void afterPropertiesSet() throws Exception {
    		new Thread(){
    			public void run() {
    				echoService.initService();
    			};
    		}.start();
    		System.out.println("netty启动成功");
    	}
    
    }

2.给普通的netty加上注解

@Component
    public class EchoService {
    	
    	@Autowired
    	private EchoServiceHandler echoServiceHandler;
    	
    	public void initService() {
    		NioEventLoopGroup nioEventLoopGroup = null;
    		try {
    			//server端引导类
    			ServerBootstrap serverBootstrap = new ServerBootstrap();
    			//连接池数据,netty自动弄的连接池
    			nioEventLoopGroup = new NioEventLoopGroup();
    			//装配bootstap
    			serverBootstrap.group(nioEventLoopGroup)
    			//通道类型为NioServerSocketChannel
    			.channel(NioServerSocketChannel.class)
    			//监听端口号
    			.localAddress(12345)
    					.childHandler(new ChannelInitializer<Channel>() {
    						@Override
    						protected void initChannel(Channel channel) throws Exception {
    							//全是模板代码，主要是这里绑定handler
    							channel.pipeline().addLast(echoServiceHandler);
    						}
    					});
    			// 最后绑定服务器等待直到绑定完成，调用sync()方法会阻塞直到服务器完成绑定,然后服务器等待通道关闭，因为使用sync()，所以关闭操作也会被阻塞。
    			ChannelFuture channelFuture = serverBootstrap.bind().sync();
    			System.out.println("netty服务端启动成功");
    			channelFuture.channel().closeFuture().sync();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}finally {
    			try {
    				nioEventLoopGroup.shutdownGracefully().sync();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    }

@Component
    @Sharable//注意这个注解
    public class EchoServiceHandler extends ChannelInboundHandlerAdapter {
    
    	@Autowired
    	private NettyHelp nettyHelp;
    	
    	/**
    	 * 服务端获取请求的时候被调用
    	 */
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    		ByteBuf buf = (ByteBuf)msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String content = new String(req,"UTF-8");
        	System.out.println("netty服务端接受到的消息是："+content);
            List<TUser> list = nettyHelp.getrsp(Integer.valueOf(content));
            String rsp = JSONArray.toJSONString(list);
            ByteBuf rspBuf = Unpooled.copiedBuffer(rsp.getBytes());
            ctx.write(rspBuf);//响应给客户端
    	}
    	
    	/**
    	 * channelRead方法完成之后调用
    	 */
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		ctx.flush();//flush之后才是真的发送了
    		System.out.println("netty服务端响应成功");
    		//ctx.close();
    	}
    	
    	/**
    	 * 发生异常时候被调用
    	 */
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		System.out.println("服务端捕获到异常，关闭ctx");
    		cause.printStackTrace();
    		ctx.close();
    	}
    	
    }

3.启动类

public class StartProject {
    
    	@SuppressWarnings("resource")
    	public static void main(String[] args) {
    		try {
    			ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:applicationContext.xml");
    			context.start();
    			System.out.println("------------------启动成功----------------------");
    		} catch (Exception e) {
    			e.printStackTrace();
    			System.out.println("------------------启动失败----------------------");
    		}
    		
    		//下面就是让这个类一直跑着,为controller提供服务
    		synchronized (StartProject.class) {
    			while (true) {
    				try {
    					StartProject.class.wait();
    				} catch (InterruptedException e) {
    					
    				}
    			}
    		}
    	}
    }