文章目录：
- 1、socket多进程并发服务器怎样实现预先创建几个进程
- 2、那用socket连接服务器的并发访问量超出了服务器的承载能力，这个要怎么解决呢？
- 3、如何用socket与多线程实现在服务器端并发处理多客户端的请求
- 4、epoll可以解决多个socket的连接，为什么高并发服务器还要用进程池或者线程池呢？
- 5、socket高并发网络编程服务端有什么框架？
socket多进程并发服务器怎样实现预先创建几个进程
Windows的网络编程我不熟悉，我只了解Linux网络编程，不过我想原理应该是互通的。
在Linux中大致思路是，由主进程完成Listen监听操作之后，就开始预先派生子进程fork操作，然后在子进程的无限循环中accpet等待客户端来连接。
int main()
{
socket();
bind();
listen();
pids[i]=your_fork_function();
for(;;)
pause();
return 0;
}
-------------------------------------------------------------------------------------------------------------
具体你可以参见《Unix网络编程.第一卷》第三十章，P712的内容。
那用socket连接服务器的并发访问量超出了服务器的承载能力，这个要怎么解决呢？
服务器socket只创一个实例就可以，然后创建监听文件描述符，然后用select或者poll 进行并发监听，来了请求就进行处理一下，数据处理完成就关掉数据联接。
但即使这样，在大规模商用的时候，还是可能超出一个服务器的能力，要负载平衡,多个服务器平均承担负载。
不过看你描述，你的程序应该还远没到大规模商用的地步。
如何用socket与多线程实现在服务器端并发处理多客户端的请求
你至少说一下你用什么语言吧？给你个伪代码，C、C++，java或其他语言的处理过程基本相似：
//创建服务端Socket，指定地址族，连接地址，传输协议
ServerSocket servSocket = new ServerSocket (AF_INET,"127.0.0.1",STREAM);
//指定服务端启用端口
bind(servSocket ,port);
//把服务端socket转化为监听socket
listene(servSocket );
//监听客户端的请求
for(;;) {
ClientSocket cliSocket =new ClientSocket ();
//此处没有连接请求的时候会产生阻塞，会把主线程挂起，有连接请求由操作系统或运行环境通知主线程，继续处理
accept(servSocket,cliSocket);
//执行到此处说明有客户端请求，创建线程处理客户端请求,此处耗费的时间仅仅是各语言的线程创建时间，不处理任何其他工作，具体工作写到线程回调代码中执行。
createNewThreadHandleClientRequet(cliSocket);
//到此循环回去，等待下一次客户端请求
}
epoll可以解决多个socket的连接，为什么高并发服务器还要用进程池或者线程池呢？
socket接受线程：C语言为了高并发所以选择了epoll。当程序启动的时候（g_net_update.c文件中main函数，会启动一个thread见函数create_accept_task）这个thread就处理一件事情，只管接收客户端的连接，当有连接进来的时候 通过epoll_ctl函数，把socket fd 加入到epoll里面去，epoll设置监听事件EPOLLIN | EPOLLET; 主要是监听的是加入到epoll中的socket是否可读(因为我的需求是客户端连上了server就会马上向server发送一份数据的)。其它的部分在主线程中处理。
主线程：是一个无线循环，epoll_wait 函数相当于把客户端的连接从epoll中拿出来（因为我们监听的是EPOLLIN | EPOLLET）说明这个时候客户端有数据发送过来）。再通过recv_buffer_from_fd 函数把客户端发送过来的数据读出来。然后其他的一切就抛给线程池去处理。
线程池：(代码中我会在池里面创建15个线程) 双向链表。加入线程就是在链表后面加一个链表项，链表的前面会一个一个被拿出来处理。主要是malloc 函数free函数，sem_wait函数sem_post的处理（sem_wait 会阻塞当值大于0是会减一，sem_post是值加一）。typedef void* (FUNC)(void arg, int index);是我们自定义的线程的逻辑处理部分，arg是参数，index是第几个线程处理(我们隐形的给每个线程都标了号)，例如代码中的respons_stb_info，更加具体可以看看代码里面是怎么实现的。聪明的你也可以改掉这块的内容改成动态线程池，当某个时刻的处理比较多的时候能够动态的增加线程，而不像我代码里面的是固定的。
数据库连接池：按照我的需求在处理客户端请求数据的时候是要访问数据库的。就是一下子创建出一堆的数据连接。要访问数据库的时候先去数据库连接池中找出空闲的连接，具体可以看下代码。使用的时候可以参考下database_process.c文件（代码中数据库连接池和线程池中的个数是一样的）。这里我想说下get_db_connect_from_pool这个函数，我用了随机数，我是为了不想每次都从0开始去判断哪个连接没有用到。为了数据库连接池中的每个链接都能等概率的使用到，具体的还是可以看下代码的实现。
socket高并发网络编程服务端有什么框架？
netty;
PayServer.java
package com.miri.pay.scoket;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class PayServer implements Runnable
{
private static final Logger DLOG = LoggerFactory.getLogger(PayServer.class);
private final int port;
public PayServer(int port)
{
this.port = port;
}
/**
* 为ServerBootstrap绑定指定端口
*/
public void run()
{
// 用于接收发来的连接请求
final EventLoopGroup bossGroup = new NioEventLoopGroup();
// 用于处理boss接受并且注册给worker的连接中的信息
final EventLoopGroup workerGroup = new NioEventLoopGroup();
try
{
// 配置服务器
final ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.option(ChannelOption.SO_BACKLOG, 128);
// 通过NoDelay禁用Nagle,使消息立即发出去，不用等待到一定的数据量才发出去
bootstrap.option(ChannelOption.TCP_NODELAY, true);
// 保持长连接状态
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
// CustomChannelInitializer是一个特殊的handler，用于方便的配置用户自定义的handler实现
bootstrap.childHandler(new CustomChannelInitializer());
// 绑定并开始接受传入的连接
final ChannelFuture future = bootstrap.bind(this.port).sync();
if (future.isSuccess())
{
PayServer.DLOG.info("Start the socket server {} success", this.port);
}
else
{
PayServer.DLOG.info("Start the socket server {} failure,System exit!", this.port);
throw new RuntimeException("Socket服务端启动失败");
}
// 等待服务器套接字关闭
// 关闭服务器
future.channel().closeFuture().sync();
}
catch (final InterruptedException e)
{
PayServer.DLOG.error("Close the socket server exception occurs,System exit!", e);
throw new RuntimeException("关闭Socket服务端失败");
}
finally
{
// 关闭所有事件循环终止线程
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
/**
* 特殊的内部类
* p
* 是一个特殊的handler，用于方便的配置用户自定义的handler实现
* @author xulonghui
*/
static class CustomChannelInitializer extends ChannelInitializerSocketChannel
{
@Override
protected void initChannel(SocketChannel ch) throws Exception
{
final ChannelPipeline p = ch.pipeline();
p.addLast(new PayMessageEncoder());
p.addLast(new PayMessageDecoder());
p.addLast(new PayServerHandler());
}
}
}
PayMessageEncoder.java
package com.miri.pay.scoket;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.util.CharsetUtil;
import com.miri.pay.model.CommonResponse;
import com.miri.pay.utils.JsonUtil;
/**
*消息编码器
* p
* 编码从服务端发送出的消息
*/
public class PayMessageEncoder extends MessageToByteEncoderCommonResponse
{
@Override
protected void encode(ChannelHandlerContext ctx, CommonResponse rsp, ByteBuf out) throws Exception
{
if (rsp != null)
{
final Object msgContent = rsp.getMsgContent();
// 消息ID,sequenceId和entityId三个加起来是12个长度
int msgLen = 12;
byte[] contentbs = new byte[] {};
if (msgContent != null)
{
final String content = JsonUtil.bean2json(msgContent);
contentbs = content.getBytes(CharsetUtil.UTF_8);
final int cl = contentbs.length;
msgLen += cl;
}
out.writeInt(msgLen);// 写入当前消息的总长度
out.writeInt(rsp.getMsgId());// 写入当前消息的消息ID
out.writeInt(rsp.getSequenceId());// 写入当前消息的SequenceId
out.writeInt(rsp.getEntityId());// 写入当前消息的EntityId
// 写入消息主体内容
if (contentbs.length 0)
{
out.writeBytes(contentbs);
}
}
}
}
PayMessageDecoder.java
package com.miri.pay.scoket;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.CharsetUtil;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.miri.pay.constants.Constants;
import com.miri.pay.model.CommonRequest;
import com.miri.pay.utils.ByteUtil;
/**
* 消息解码器
* p
* 解码从客户端请求的消息
*/
public class PayMessageDecoder extends ByteToMessageDecoder
{
private static final Logger DLOG = LoggerFactory.getLogger(PayMessageDecoder.class);
/**
* 表示头长度的字节数
*/
private static final int HEAD_LENGTH = 4;
/**
* 所有ID串所属的字节数
*/
private static final int ID_STR_LENGTH = 12;
/**
* 单个ID所属的字节数
*/
private static final int SINGLE_ID_LENGTH = 4;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, ListObject out) throws Exception
{
int readable = in.readableBytes();
if (readable PayMessageDecoder.HEAD_LENGTH)
{
return;
}
in.markReaderIndex(); // 我们标记一下当前的readIndex的位置
final int dataLength = in.readInt(); // 读取传送过来的消息的长度。ByteBuf 的readInt()方法会让他的readIndex增加4
if (dataLength 0)
{
// 我们读到的消息体长度为0，这是不应该出现的情况，这里出现这情况，关闭连接。
ctx.close();
}
readable = in.readableBytes();
if (readable dataLength)
{
// 读到的消息体长度如果小于我们传送过来的消息长度，则resetReaderIndex. 这个配合markReaderIndex使用的。把readIndex重置到mark的地方
in.resetReaderIndex();
return;
}
final byte[] body = new byte[dataLength];
in.readBytes(body);
// 判断是否读取到内容
final int length = body.length;
if (length == 0)
{
return;// 若未读出任何内容，则忽略
}
out.add(this.byte2req(body));
}
/**
* 将读取到的byte数据转换为请求对象
* @param body
* @return
* @throws Exception
*/
private CommonRequest byte2req(byte[] body) throws Exception
{
final CommonRequest req = new CommonRequest(Constants.INVALID_MSGID);
final int length = body.length;
// 若内容数组的长度小于或等于12，则表示消息主体内容为空，直接返回一个无效的消息出去
if (length PayMessageDecoder.ID_STR_LENGTH)
{
PayMessageDecoder.DLOG
.info("The client sends the message length is: {}, is invalid message, directly returns a msgId = {} request entity",
length, Constants.INVALID_MSGID);
return req;
}
// 获取消息ID
final byte[] mbs = new byte[PayMessageDecoder.SINGLE_ID_LENGTH];
System.arraycopy(body, 0, mbs, 0, PayMessageDecoder.SINGLE_ID_LENGTH);
final int msgId = ByteUtil.byte4toint(mbs);
req.setMsgId(msgId);
// 获取sequenceId
final byte[] sbs = new byte[PayMessageDecoder.SINGLE_ID_LENGTH];
System.arraycopy(body, 4, sbs, 0, PayMessageDecoder.SINGLE_ID_LENGTH);
final int sequenceId = ByteUtil.byte4toint(sbs);
req.setSequenceId(sequenceId);
// 获取entityId
final byte[] ebs = new byte[PayMessageDecoder.SINGLE_ID_LENGTH];
System.arraycopy(body, 8, ebs, 0, PayMessageDecoder.SINGLE_ID_LENGTH);
final int entityId = ByteUtil.byte4toint(ebs);
req.setEntityId(entityId);
// 获取消息主体内容
if (length PayMessageDecoder.ID_STR_LENGTH)
{
final int contentLen = length - PayMessageDecoder.ID_STR_LENGTH;
final byte[] contentbs = new byte[contentLen];
System.arraycopy(body, 12, contentbs, 0, contentLen);
final String content = new String(contentbs, CharsetUtil.UTF_8);
req.setMsgContent(content);
}
return req;
}
}
PayServerHandler.java
package com.miri.pay.scoket;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import java.util.HashMap;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.miri.pay.MessageQueue;
import com.miri.pay.model.CommonRequest;
import com.miri.pay.model.PendingBean;
/**
* Socket服务端处理器
*/
public class PayServerHandler extends ChannelInboundHandlerAdapter
{
private static final Logger DLOG = LoggerFactory.getLogger(PayServerHandler.class);
/**
* 外部订单号-频道
*/
public static final MapString, Channel CHANNELS = new HashMapString, Channel();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
{
try
{
PayServerHandler.DLOG.info("Client send to msg is: {}", msg);
final CommonRequest request = (CommonRequest) msg;
final PendingBean bean = new PendingBean(ctx.channel(), request);
MessageQueue.offer(bean);
}
finally
{
ReferenceCountUtil.release(msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception
{
ctx.flush();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception
{
super.channelActive(ctx);
final Channel channel = ctx.channel();
PayServerHandler.DLOG.info("Client active form {}", channel.remoteAddress());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception
{
super.channelInactive(ctx);
final Channel channel = ctx.channel();
PayServerHandler.DLOG.info("Client inactive form {}", channel.remoteAddress());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
{
PayServerHandler.DLOG.error("System exception", cause);
ctx.close();
}
}
代码中执行。createNewThreadHandleClientRequet(cliSocket);//到此循环回去，等待下一次客户端请求}epoll可以解决多个socket的连接，为什么高并发服务器还要用进程池或者线程池呢
LENGTH) { return; } in.markReaderIndex(); // 我们标记一下当前的readIndex的位置 final int dataL
监听客户端的请求for(;;) {ClientSocket cliSocket =new ClientSocket ();//此处没有连接请求的时候会产生阻塞，会把主线程挂起，有连接请求由操作系统或运行环境通知主线程，继续处理accept(servSocke
ss()) { PayServer.DLOG.info("Start the socket server {} success", this.port); } else { PayServer.DLOG.info("Start the
ServerSocket servSocket = new ServerSocket (AF_INET,"127.0.0.1",STREAM);//指定服务端启用端口bind(servSocket ,port);//把服