Netty实现一个基于socket传输程序

概述

  上一篇博客介绍了使用netty来实现一个简单的http服务程序。HTTP协议是一个无状态的、短连接的，基于请求响应的一个协议。所谓的无状态，指的是前后两次的请求是毫不相关的，也就是说后一次请求对于前一次请求的数据是完全未知的。所以才会出现使用session、cookie来进行一些数据的存储。所谓的短连接，指的是一次请求响应完成后，链接就会断开（当然，这个对于HTTP 1.1版本不一定正确。这个需要根据header里面的keep-alive字段来决定要不要立即断开连接）。所谓的基于请求响应，是指由客户端那边发送请求，服务端这边收到请求过后给出响应发送给客户端，完成一次请求响应。也就是说，在http的规范下，只能有客户端主动向服务端发送数据，而不能由服务端向客户端主动的推送数据。如果出现了这样的需求，就只能采用轮训的方式来查询服务端状态在做出对应处理。
  而基于socket编程的TCP协议则不一样。TCP协议是一个面向连接的一个传输层协议。所谓的面向连接，是指在传输数据之前，客户端和服务端之间首先要建立连接（三次握手）。一旦连接建立成功，就可以进行数据传输了。连接建立后服务端和客户端是以一种全双工的方式来进行数据传输的，也就是说客户端可以主动向服务端发送数据，反过来服务端也可以反过来想客户端发送数据。而且连接建立好了后，服务端或者客户端发送数据后，这个连接不是就断开了，还可以继续的发送数据。这个跟HTTP协议的规范显然是大不相同的。但是HTTP却是构建与TCP协议之上的一个应用层协议，其底层还是基于TCP协议来实现的。只不过HTTP加了很多自己的规范。
  以上就是HTTP协议和基于socket编程的TCP协议的区别和联系。本篇博客主要来介绍下怎么使用Netty来实现一个基于TCP的socket编程。

运行环境

mac pro

Itellij Idea

gradle

jdk 8

程序所需jar包

sourceCompatibility = 1.8
targetCompatibility = 1.8

repositories {
mavenCentral()
}

dependencies {
testCompile (
"junit:junit:4.12",
"io.netty:netty-all:4.1.11.Final"
)
}

netty客户端开发的一般流程

  上一篇博客介绍了netty服务端开发的一般流程，其实客户端跟服务端类似。一下是netty客户端开发的一般流程：

编写客户端main方法，其中定义一个事件驱动循环组对象 NioEventLoopGroup（服务端是两个事件循环组，一个用来处理链接，一个用来处理逻辑，客户端只需要处理链接，因此只要一个循环组就可以了），然后定义一个客户端启动类Bootstrap对象，设置通道（NioSocketChannel.class）并且设置Handler对象（这里解释下，handler对应的是bossGroup,而childHandler对应的是workerGroup），也就是第二步定义的Initializer类对应的对象，连接到对应的地址端口。

新建一个Initializer类，该类一般继承ChannelInitializer

<

SocketChannel

>

类。在initChannel方法里面添加对应的handler。

新建一个我们自己的handler，在channelRead0方法里面编写从客户端读取消息的处理逻辑。该方法的第二个参数就是从客户端读取的消息。在本篇博客里面，实现的功能是客户端和服务端建立好链接过后，客户端向服务端发送一条消息，然后服务端吧收到的消息在发送给客户端。因此需要在handler里面增加channelActive方法，用于链接建立好了过后向服务端发送一个数据。

Netty实现一个基于socket传输程序代码

服务端代码

package com.shengsiyuan.netty.secondexample;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

/** * Socket编程服务端 * @Author: zhouwen * @Date: 2017/6/10 13:14 */
public class MyServer {

public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new MyServerInitializer());

ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

package com.shengsiyuan.netty.secondexample;

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

/** * 服务端Initializer * @Author: zhouwen * @Date: 2017/6/10 13:20 */
public class MyServerInitializer extends ChannelInitializer<SocketChannel> {

@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//编解码的处理
pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast(new LengthFieldPrepender(4));
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
//自己的处理器
pipeline.addLast(new MyServerHandler());
}
}

package com.shengsiyuan.netty.secondexample;

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

import java.util.UUID;

/** * 自己定义的一个handler * @Author: zhouwen * @Date: 2017/6/10 13:25 */
public class MyServerHandler extends SimpleChannelInboundHandler<String> {

@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(ctx.channel().remoteAddress() + ", " + msg);

ctx.channel().writeAndFlush("from server: " + UUID.randomUUID());
}

/** * 对于异常的处理 * @param ctx * @param cause * @throws Exception */
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

客户端代码

package com.shengsiyuan.netty.secondexample;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

/** * socket编程的客户端 * @Author: zhouwen * @Date: 2017/6/10 13:49 */
public class MyClient {

public static void main(String[] args) throws Exception {
// 客户端只有一个事件循环组
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
// 客户端启动器
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new MyClientInitializer());

ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8899).sync();
channelFuture.channel().closeFuture().sync();
} finally {
eventLoopGroup.shutdownGracefully();
}
}
}

package com.shengsiyuan.netty.secondexample;

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

/** * 客户端的Initializer * @Author: zhouwen * @Date: 2017/6/10 13:54 */
public class MyClientInitializer extends ChannelInitializer<SocketChannel> {

@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//编解码的处理
pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast(new LengthFieldPrepender(4));
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
//自己的处理器
pipeline.addLast(new MyClientHandler());
}
}

package com.shengsiyuan.netty.secondexample;

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

import java.time.LocalDateTime;

/** * 客户端处理器 * @Author: zhouwen * @Date: 2017/6/10 13:56 */
public class MyClientHandler extends SimpleChannelInboundHandler<String> {

@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(ctx.channel().remoteAddress());
System.out.println("client output: " + msg);

ctx.writeAndFlush("from client: " + LocalDateTime.now());
}

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

/** * channel处于活动状态后的回调（链接建立好了过后） * @param ctx * @throws Exception */
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.channel().writeAndFlush("来与客户端的问候！");
}
}

测试工作

首先启动服务器，然后再启动两个客户端 分别看看控制台输出。

服务端控制台：



第一个客户端控制台：



第二个客户端控制台



从以上可以看出，服务端启动后，客户端启动时会与服务端建立连接，简历好了过后会服务端发送一条数据“来自客户端的问候”,然后服务端会向客户端发送一个随机的UUID，客户端收到这个UUID过后，会把当前时间有发送给服务端，如此一直发送数据。以上就是netty实现一个简单的基于socket传输的程序。