Netty实战与调优
聊天室业务介绍
代码参考
/**
* 用户管理接口
*/
public interface UserService {
/**
* 登录
* @param username 用户名
* @param password 密码
* @return 登录成功返回 true, 否则返回 false
*/
boolean login(String username, String password);
}
/**
* 会话管理接口
*/
public interface Session {
/**
* 绑定会话
* @param channel 哪个 channel 要绑定会话
* @param username 会话绑定用户
*/
void bind(Channel channel, String username);
/**
* 解绑会话
* @param channel 哪个 channel 要解绑会话
*/
void unbind(Channel channel);
/**
* 获取属性
* @param channel 哪个 channel
* @param name 属性名
* @return 属性值
*/
Object getAttribute(Channel channel, String name);
/**
* 设置属性
* @param channel 哪个 channel
* @param name 属性名
* @param value 属性值
*/
void setAttribute(Channel channel, String name, Object value);
/**
* 根据用户名获取 channel
* @param username 用户名
* @return channel
*/
Channel getChannel(String username);
}
/**
* 聊天组会话管理接口
*/
public interface GroupSession {
/**
* 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null
* @param name 组名
* @param members 成员
* @return 成功时返回组对象, 失败返回 null
*/
Group createGroup(String name, Set<String> members);
/**
* 加入聊天组
* @param name 组名
* @param member 成员名
* @return 如果组不存在返回 null, 否则返回组对象
*/
Group joinMember(String name, String member);
/**
* 移除组成员
* @param name 组名
* @param member 成员名
* @return 如果组不存在返回 null, 否则返回组对象
*/
Group removeMember(String name, String member);
/**
* 移除聊天组
* @param name 组名
* @return 如果组不存在返回 null, 否则返回组对象
*/
Group removeGroup(String name);
/**
* 获取组成员
* @param name 组名
* @return 成员集合, 没有成员会返回 empty set
*/
Set<String> getMembers(String name);
/**
* 获取组成员的 channel 集合, 只有在线的 channel 才会返回
* @param name 组名
* @return 成员 channel 集合
*/
List<Channel> getMembersChannel(String name);
}
注册事件处理器
@Slf4j
public class ChatServer {
public static void main(String[] args) {
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup();
LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
LoginRequestMessageHandler LOGIN_HANDLER = new LoginRequestMessageHandler();
ChatRequestMessageHandler CHAT_HANDLER = new ChatRequestMessageHandler();
GroupCreateRequestMessageHandler GROUP_CREATE_HANDLER = new GroupCreateRequestMessageHandler();
GroupChatMessageHandler GROUP_CHAT_HANDLER = new GroupChatMessageHandler();
QuitHandler QUIT_HANDLER = new QuitHandler();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//空闲检测 arg1:读空闲 arg2:写空闲 arg3:读写空闲 单位:s 0代表不关注
ch.pipeline().addLast(new IdleStateHandler(5,0,0));
//处理空闲事件
ch.pipeline().addLast(new ChannelDuplexHandler(){
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.READER_IDLE)){
log.info("读空闲超过5s!");
}
}
}
});
ch.pipeline().addLast(new ProcotolFrameDecoder());
ch.pipeline().addLast(LOGGING_HANDLER);
ch.pipeline().addLast(MESSAGE_CODEC);
ch.pipeline().addLast(LOGIN_HANDLER);
ch.pipeline().addLast(CHAT_HANDLER);
ch.pipeline().addLast(GROUP_CREATE_HANDLER);
ch.pipeline().addLast(GROUP_CHAT_HANDLER);
ch.pipeline().addLast(QUIT_HANDLER);
}
});
Channel channel = serverBootstrap.bind(8080).sync().channel();
channel.closeFuture().sync();
} catch (InterruptedException e) {
log.error("server error", e);
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
}
}
}
客户端
@Slf4j
public class ChatClient {
public static void main(String[] args) {
NioEventLoopGroup group = new NioEventLoopGroup();
LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
AtomicBoolean LOGIN = new AtomicBoolean(false);
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(group);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ProcotolFrameDecoder());
// ch.pipeline().addLast(LOGGING_HANDLER);
ch.pipeline().addLast(MESSAGE_CODEC);
//添加心跳机制
ch.pipeline().addLast(new IdleStateHandler(0,3,0));
ch.pipeline().addLast(new ChannelDuplexHandler(){
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent event = (IdleStateEvent) evt;
//超过3s没有写入数据则触发WRITER_IDLE事件
if (IdleState.WRITER_IDLE.equals(event.state())){
// log.info("发送心跳数据包...");
ctx.writeAndFlush(new PingMessage());
}
}
}
});
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
new Thread(() -> {
Scanner scanner = new Scanner(System.in);
System.out.print("username:");
String username = scanner.nextLine();
System.out.print("password:");
String password = scanner.nextLine();
LoginRequestMessage request = new LoginRequestMessage(username, password);
ctx.channel().writeAndFlush(request);
//阻塞
try {
WAIT_FOR_LOGIN.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
if (LOGIN.get()) {
while (true) {
System.out.println("==================================");
System.out.println("send [username] [content]");
System.out.println("gsend [group name] [content]");
System.out.println("gcreate [group name] [m1,m2,m3...]");
System.out.println("gmembers [group name]");
System.out.println("gjoin [group name]");
System.out.println("gquit [group name]");
System.out.println("quit");
System.out.println("==================================");
String command = scanner.nextLine();
if (command != null && command.length() > 0) {
String[] commandList = command.split(" ");
switch (commandList[0]) {
case "send":
ChatRequestMessage message = new ChatRequestMessage(username, commandList[1], commandList[2]);
ctx.channel().writeAndFlush(message);
break;
case "gsend":
GroupChatRequestMessage groupMessage = new GroupChatRequestMessage(username, commandList[1], commandList[2]);
ctx.channel().writeAndFlush(groupMessage);
break;
case "gcreate":
String members = commandList[2];
if (members != null && members.length() > 0) {
String[] memberList = members.split(",");
Set<String> memberSet = Arrays.stream(memberList).collect(Collectors.toSet());
memberSet.add(username);
GroupCreateRequestMessage groupCreateRequestMessage = new GroupCreateRequestMessage(commandList[1], memberSet);
ctx.channel().writeAndFlush(groupCreateRequestMessage);
}
break;
case "gmembers":
GroupMembersRequestMessage groupMembersRequestMessage = new GroupMembersRequestMessage(commandList[1]);
ctx.channel().writeAndFlush(groupMembersRequestMessage);
break;
case "gjoin":
GroupJoinRequestMessage groupJoinRequestMessage = new GroupJoinRequestMessage(username, commandList[1]);
ctx.channel().writeAndFlush(groupJoinRequestMessage);
break;
case "gquit":
GroupQuitRequestMessage groupQuitRequestMessage = new GroupQuitRequestMessage(username, commandList[1]);
ctx.channel().writeAndFlush(groupQuitRequestMessage);
break;
case "quit":
ctx.channel().close();
return;
}
}
}
} else {
ctx.channel().close();
}
}).start();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("msg:{}", msg);
if (msg instanceof LoginResponseMessage response) {
LOGIN.set(response.isSuccess());
WAIT_FOR_LOGIN.countDown();
}
}
});
}
});
Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
channel.closeFuture().sync();
} catch (Exception e) {
log.error("client error", e);
} finally {
group.shutdownGracefully();
}
}
}
登录
@ChannelHandler.Sharable
public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
String username = msg.getUsername();
String password = msg.getPassword();
boolean login = UserServiceFactory.getUserService().login(username, password);
if (login) {
SessionFactory.getSession().bind(ctx.channel(), username);
}
ctx.channel().writeAndFlush(new LoginResponseMessage(login, login ? "登录成功!" : "登录失败,用户名或密码有误!"));
}
}
单聊
@ChannelHandler.Sharable
public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, ChatRequestMessage chatRequestMessage) throws Exception {
String to = chatRequestMessage.getTo();
String from = chatRequestMessage.getFrom();
String content = chatRequestMessage.getContent();
Channel channel = SessionFactory.getSession().getChannel(to);
if (channel!=null){
//转发消息
channel.writeAndFlush(new ChatResponseMessage(from,content));
}
else {
channelHandlerContext.writeAndFlush(new ChatResponseMessage(false,"发送失败,用户不在线!"));
}
}
}
创建群聊
@ChannelHandler.Sharable
@Slf4j
public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, GroupCreateRequestMessage groupCreateRequestMessage) throws Exception {
Set<String> members = groupCreateRequestMessage.getMembers();
String groupName = groupCreateRequestMessage.getGroupName();
GroupSession session = GroupSessionFactory.getGroupSession();
Group group = session.createGroup(groupName, members);
GroupCreateResponseMessage message = null;
if (group != null) {
message = new GroupCreateResponseMessage(false, "创建群聊失败!");
} else {
message = new GroupCreateResponseMessage(true, "创建群聊成功!");
List<Channel> channels = session.getMembersChannel(groupName);
channels.forEach(channel -> {
GroupJoinResponseMessage joinResponseMessage = new GroupJoinResponseMessage(true, "您已被拉入群聊:" + groupName);
channel.writeAndFlush(joinResponseMessage);
});
}
channelHandlerContext.writeAndFlush(message);
}
}
发送群聊消息
@ChannelHandler.Sharable
public class GroupChatMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, GroupChatRequestMessage groupChatRequestMessage) throws Exception {
String groupName = groupChatRequestMessage.getGroupName();
String msg = groupChatRequestMessage.getContent();
String from = groupChatRequestMessage.getFrom();
List<Channel> channels = GroupSessionFactory.getGroupSession().getMembersChannel(groupName);
channels.forEach(channel -> {
channel.writeAndFlush(new GroupChatResponseMessage(from, msg));
});
}
}
退出
@ChannelHandler.Sharable
@Slf4j
public class QuitHandler extends ChannelInboundHandlerAdapter {
//处理channel正常关闭事件
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
//移除channel
SessionFactory.getSession().unbind(ctx.channel());
log.info("channel:{}已断开!",ctx.channel());
}
//处理channel异常关闭事件
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
SessionFactory.getSession().unbind(ctx.channel());
log.info("channel:{}异常断开!",ctx.channel());
}
}
空闲检测
连接假死
原因
- 网络设备出现故障，例如网卡，机房等，底层的 TCP 连接已经断开了，但应用程序没有感知到，仍然占用着资源。
- 公网网络不稳定，出现丢包。如果连续出现丢包，这时现象就是客户端数据发不出去，服务端也一直收不到数据，就这么一直耗着
- 应用程序线程阻塞，无法进行数据读写
问题
- 假死的连接占用的资源不能自动释放
- 向假死的连接发送数据，得到的反馈是发送超时
服务器端解决
- 怎么判断客户端连接是否假死呢？如果能收到客户端数据，说明没有假死。因此策略就可以定为，每隔一段时间就检查这段时间内是否接收到客户端数据，没有就可以判定为连接假死
//空闲检测 arg1:读空闲 arg2:写空闲 arg3:读写空闲 单位:s 0代表不关注
ch.pipeline().addLast(new IdleStateHandler(5,0,0));
//处理空闲事件
ch.pipeline().addLast(new ChannelDuplexHandler(){
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.READER_IDLE)){
log.info("读空闲超过5s!");
//关闭channel
}
}
}
});
客户端定时心跳
- 客户端可以定时向服务器端发送数据，只要这个时间间隔小于服务器定义的空闲检测的时间间隔，那么就能防止前面提到的误判，客户端可以定义如下心跳处理器
//添加心跳机制
ch.pipeline().addLast(new IdleStateHandler(0,3,0));
ch.pipeline().addLast(new ChannelDuplexHandler(){
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent event = (IdleStateEvent) evt;
//超过3s没有写入数据则触发WRITER_IDLE事件
if (IdleState.WRITER_IDLE.equals(event.state())){
log.info("发送心跳数据包...");
ctx.writeAndFlush(new PingMessage());
}
}
}
});
优化
拓展序列化算法
序列化，反序列化主要用在消息正文的转换上
- 序列化时，需要将 Java 对象变为要传输的数据（可以是 byte[]，或 json 等，最终都需要变成 byte[]）
- 反序列化时，需要将传入的正文数据还原成 Java 对象，便于处理
为了支持更多序列化算法，抽象一个 Serializer 接口
public interface Serializer {
// 反序列化方法
<T> T deserialize(Class<T> clazz, byte[] bytes);
// 序列化方法
<T> byte[] serialize(T object);
}
提供两个实现，我这里直接将实现加入了枚举类 Serializer.Algorithm 中
enum Algorithm implements Serializer {
Java {
@Override
public <T> T deserialize(Class<T> clazz, byte[] bytes) {
try {
ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes));
Object object = in.readObject();
return (T) object;
} catch (IOException | ClassNotFoundException e) {
throw new RuntimeException("SerializerAlgorithm.Java 反序列化错误", e);
}
}
@Override
public <T> byte[] serialize(T object) {
try {
ByteArrayOutputStream out = new ByteArrayOutputStream();
new ObjectOutputStream(out).writeObject(object);
return out.toByteArray();
} catch (IOException e) {
throw new RuntimeException("SerializerAlgorithm.Java 序列化错误", e);
}
}
},
// Json 实现(引入了 Gson 依赖)
Json {
@Override
public <T> T deserialize(Class<T> clazz, byte[] bytes) {
return new Gson().fromJson(new String(bytes, StandardCharsets.UTF_8), clazz);
}
@Override
public <T> byte[] serialize(T object) {
return new Gson().toJson(object).getBytes(StandardCharsets.UTF_8);
}
};
// 需要从协议的字节中得到是哪种序列化算法
public static Algorithm getByInt(int type) {
Algorithm[] array = Algorithm.values();
if (type < 0 || type > array.length - 1) {
throw new IllegalArgumentException("Out of index!");
}
return array[type];
}
}
增加配置类和配置文件
public abstract class Config {
static Properties properties;
static {
try (InputStream in = Config.class.getResourceAsStream("/application.properties")) {
properties = new Properties();
properties.load(in);
} catch (IOException e) {
throw new ExceptionInInitializerError(e);
}
}
public static int getServerPort() {
String value = properties.getProperty("server.port");
if(value == null) {
return 8080;
} else {
return Integer.parseInt(value);
}
}
public static Serializer.Algorithm getSerializerAlgorithm() {
String value = properties.getProperty("serializer.algorithm");
if(value == null) {
return Serializer.Algorithm.Java;
} else {
return Serializer.Algorithm.valueOf(value);
}
}
}
配置文件
serializer.algorithm=Json
修改编解码器
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
@Override
protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
ByteBuf out = ctx.alloc().buffer();
// 1. 4 字节的魔数
out.writeBytes(new byte[]{1, 2, 3, 4});
// 2. 1 字节的版本,
out.writeByte(1);
// 3. 1 字节的序列化方式 jdk 0 , json 1
out.writeByte(0);
// 4. 1 字节的指令类型
out.writeByte(msg.getMessageType());
// 5. 4 个字节
out.writeInt(msg.getSequenceId());
// 无意义，对齐填充
out.writeByte(0xff);
// 6. 获取内容的字节数组
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(msg);
byte[] bytes = bos.toByteArray();
// 7. 长度
out.writeInt(bytes.length);
// 8. 写入内容
out.writeBytes(bytes);
outList.add(out);
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
int magicNum = in.readInt();
byte version = in.readByte();
byte serializerType = in.readByte();
byte messageType = in.readByte();
int sequenceId = in.readInt();
in.readByte();
int length = in.readInt();
byte[] bytes = new byte[length];
in.readBytes(bytes, 0, length);
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
Message message = (Message) ois.readObject();
log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
log.debug("{}", message);
out.add(message);
}
}
参数调优
CONNECT_TIMEOUT_MILLIS
- 属于 SocketChannal的参数
- 用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常
- 注意：Netty 中不要用成了SO_TIMEOUT 主要用在阻塞 IO，而 Netty 是非阻塞 IO
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,300);// SocketChannel0.3s内未建立连接就抛出异常
源码部分 io.netty.channel.nio.AbstractNioChannel.AbstractNioUnsafe#connect
// Schedule connect timeout.
int connectTimeoutMillis = config().getConnectTimeoutMillis();//获取超时时间
if (connectTimeoutMillis > 0) {
//创建定时任务
connectTimeoutFuture = eventLoop().schedule(new Runnable() {
@Override
public void run() { //如果超时后Promise依然存在,则向Promise标记异常
ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
ConnectTimeoutException cause =
new ConnectTimeoutException("connection timed out: " + remoteAddress);
if (connectPromise != null && connectPromise.tryFailure(cause)) {
close(voidPromise());
}
}
}, connectTimeoutMillis, TimeUnit.MILLISECONDS);//在connectTimeoutMillis后开始执行任务
}
SO_BACKLOG
- 该参数是ServerSocketChannel的参数
三次握手与连接队列
简单易懂的解释
1.第一次握手时，因为客户端与服务器之间的连接还未完全建立，连接会被放入半连接队列中
2.当完成三次握手以后，连接会被放入全连接队列中
3.服务器处理Accept事件是在TCP三次握手，也就是建立连接之后。服务器会从全连接队列中获取连接并进行处理
详细的解释
- 第一次握手，client 发送 SYN 到 server，状态修改为 SYN_SEND，server 收到，状态改变为 SYN_REVD，并将该请求放入 sync queue 队列
- 第二次握手，server 回复 SYN + ACK 给 client，client 收到，状态改变为 ESTABLISHED，并发送 ACK 给 server
- 第三次握手，server 收到 ACK，状态改变为 ESTABLISHED，将该请求从 sync queue 放入 accept queue
Linux的backlog 参数
在 Linux 2.2 之前，backlog 大小包括了两个队列的大小，在 Linux 2.2 之后，分别用下面两个参数来控制
- 半连接队列 - sync queue
- 大小通过
/proc/sys/net/ipv4/tcp_max_syn_backlog指定，在syncookies启用的情况下，逻辑上没有最大值限制，这个设置便被忽略- 全连接队列 - accept queue
- 其大小通过
/proc/sys/net/core/somaxconn指定，在使用 listen 函数时，内核会根据传入的 backlog 参数与系统参数，取二者的较小值- 如果 accept queue 队列满了，server 将发送一个拒绝连接的错误信息到 client
SO_BACKLOG作用
在Netty中，SO_BACKLOG主要用于设置全连接队列的大小。当处理Accept的速率小于连接建立的速率时，全连接队列中堆积的连接数大于SO_BACKLOG设置的值时，便会抛出异常
设置方式
// 设置全连接队列，大小为2
new ServerBootstrap().option(ChannelOption.SO_BACKLOG, 2);
SO_BACKLOG默认值
查询方法调用ServerSocketChannelImpl.bind
@Override
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
synchronized (stateLock) {
ensureOpen();
if (localAddress != null)
throw new AlreadyBoundException();
if (isUnixSocket()) {
localAddress = unixBind(local, backlog);
} else {
localAddress = netBind(local, backlog);
}
}
return this;
}
在方法NioServerSocketChannel.doBind方法中调用了bind方法绑定backlog,此处调用了config.getBacklog方法
@Override
protected void doBind(SocketAddress localAddress) throws Exception {
if (PlatformDependent.javaVersion() >= 7) {
javaChannel().bind(localAddress, config.getBacklog());
} else {
javaChannel().socket().bind(localAddress, config.getBacklog());
}
}
查询config实现类
private final class NioServerSocketChannelConfig extends DefaultServerSocketChannelConfig
在父类DefaultServerSocketChannelConfig中可以看到backlog被赋值
public class DefaultServerSocketChannelConfig extends DefaultChannelConfig implements ServerSocketChannelConfig {
protected final ServerSocket javaSocket;
private volatile int backlog = NetUtil.SOMAXCONN;//为backlog赋值
}
具体赋值操作
SOMAXCONN = AccessController.doPrivileged(new PrivilegedAction<Integer>() {
@Override
public Integer run() {
// Determine the default somaxconn (server socket backlog) value of the platform.
// The known defaults:
// - Windows NT Server 4.0+: 200
// - Linux and Mac OS X: 128
int somaxconn = PlatformDependent.isWindows() ? 200 : 128;
File file = new File("/proc/sys/net/core/somaxconn");
BufferedReader in = null;
try {
// file.exists() may throw a SecurityException if a SecurityManager is used, so execute it in the
// try / catch block.
// See https://github.com/netty/netty/issues/4936
if (file.exists()) {
in = new BufferedReader(new FileReader(file));
// 将somaxconn设置为Linux配置文件中设置的值
somaxconn = Integer.parseInt(in.readLine());
if (logger.isDebugEnabled()) {
logger.debug("{}: {}", file, somaxconn);
}
} else {
...
}
...
}
// 返回backlog的值
return somaxconn;
}
}
-
backlog的值会根据操作系统的不同，来
选择不同的默认值
- Windows
200 - Linux/Mac OS
128
- Windows
-
如果配置文件
/proc/sys/net/core/somaxconn存在，会读取配置文件中的值，并将backlog的值设置为配置文件中指定的- 注意：windows系统中并不存在该文件
ulimit -n
- 属于操作系统参数
- 最大文件限制数
- 在linux下一切皆文件，开启一个进程就是打开一个文件，控制ulimit大小作用等同于限制进程及其子进程的资源使用
使用ulimit -a 可以查看当前系统的所有限制值，使用ulimit -n 可以查看当前的最大打开文件数。
新装的
linux默认只有1024，当作负载较大的服务器时，很容易遇到error: too many open files。因此，需要将其改大。使用
ulimit -n 65535可即时修改，但重启后就无效了。（注ulimit -SHn 65535 等效 ulimit -n 65535，-S指soft，-H指hard)设置永久生效
可以修改配置文件/etc/profile
vi /etc/profile加入一行
ulimit -SHn 65536保存退出。然后加载配置文件
source /etc/profile再次查看ulimit
ulimit -n第二种修改方法
在/etc/security/limits.conf最后增加如下两行记录
soft nofile 65535 hard nofile 65535
TCP_NODELAY
- 属于SocketChannal参数
- 因为Nagle算法，数据包会堆积到一定的数量后一起发送，这就可能导致
数据的发送存在一定的延时 - 该参数默认为false,如果不希望的发送被延时，则需要将该值设置为true
在有些网络通信的场景下，要求低延迟，这样就需要我们设置一些TCP的链接属性
客户端设置
bootstap.option(ChannelOption.TCP_NODELAY, true);
服务器端是在worker的Channel端设置属性
boot.childOption(ChannelOption.TCP_NODELAY, true);
设置这样做好的好处就是禁用nagle算法
-
Nagle算法试图减少TCP包的数量和结构性开销, 将多个较小的包组合成较大的包进行发送.但这不是重点, 关键是这个算法受TCP延迟确认影响, 会导致相继两次向连接发送请求包,
-
读数据时会有一个最多达500毫秒的延时.
-
TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。（一个连接会设置MSS参数，因此，TCP/IP希望每次都能够以MSS尺寸的数据块来发送数据)
-
Nagle算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块
SO_SNDBUF & SO_RCVBUF
- SO_SNDBUF 属于SocketChannal参数
- SO_RCVBUF既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）
- 该参数用于指定接收方与发送方的滑动窗口大小
- 一般不建议设置该参数，系统会自动设置大小
ALLOCATOR
- 属于SocketChannal参数
- 用来配置ByteBuf是池化还是非池化，是直接内存还是堆内存
使用
// 选择ALLOCATOR参数，设置SocketChannel中分配的ByteBuf类型
// 第二个参数需要传入一个ByteBufAllocator，用于指定生成的 ByteBuf 的类型
new ServerBootstrap().childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator());
ByteBufAllocator类型
-
池化并使用直接内存
// true表示使用直接内存 new PooledByteBufAllocator(true); -
池化并使用堆内存
// false表示使用堆内存 new PooledByteBufAllocator(false); -
非池化并使用直接内存
// ture表示使用直接内存 new UnpooledByteBufAllocator(true); -
非池化并使用堆内存
// false表示使用堆内存 new UnpooledByteBufAllocator(false);
RCVBUF_ALLOCATOR
- 属于SocketChannal参数
- 控制Netty接收缓冲区大小
- 负责入站数据的分配，决定入站缓冲区的大小（并可动态调整），统一采用 direct 直接内存，具体池化还是非池化由 allocator决定
