在第一章笔者提到在NioEndpoint类中会使用五个内部类来处理请求。在之前的文章中为了学习的方便只使用了四个类,在本章中就让我们使用 SocketProcessor处理请求。
本章程序目录结构
第一步:新建SocketProcessorBase类
public abstract class SocketProcessorBase<S> implements Runnable { protected SocketWrapperBase<S> socketWrapper; protected SocketEvent event; public SocketProcessorBase(SocketWrapperBase<S> socketWrapper, SocketEvent event) { reset(socketWrapper, event); } public void reset(SocketWrapperBase<S> socketWrapper, SocketEvent event) { Objects.requireNonNull(event); this.socketWrapper = socketWrapper; this.event = event; } @Override public final void run() { synchronized (socketWrapper) { // It is possible that processing may be triggered for read and // write at the same time. The sync above makes sure that processing // does not occur in parallel. The test below ensures that if the // first event to be processed results in the socket being closed, // the subsequent events are not processed. if (socketWrapper.isClosed()) { return; } doRun(); } } protected abstract void doRun(); }
第二步:在NioEndPoint类中新建SocketProcessor类并继承SocketProcessorBase类。在SocketProcessor类中实现doRun方法
protected class SocketProcessor extends SocketProcessorBase<NioChannel> { public SocketProcessor(SocketWrapperBase<NioChannel> socketWrapper, SocketEvent event) { super(socketWrapper, event); } @Override protected void doRun() { NioChannel socket = socketWrapper.getSocket(); SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector()); try { int handshake = -1; if (key != null) { handshake = 0; } if (handshake == 0) { Handler.SocketState state = Handler.SocketState.OPEN; if (event == null) { state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ); } else { state = getHandler().process(socketWrapper, event); } if (state == Handler.SocketState.CLOSED) { //close(socket, key); } } } catch (Exception e) { // ignore } finally { socketWrapper = null; event = null; } } }
第三步:新建AbstractEndpoint类,让NioEndpoint类继承AbstractEndpoint类。在AbstractEndpoint中定义了一个Handler接口,在这接口中
的process方法是我们当下关注的重点。AbstractEndpoint定义了一个processorCache栈属性,processorCache用来保存SocketProcessorBase
,SocketProcessorBase是用来处理请求的,当其处理完请求并不会被销毁而是保存在processorCache中以减少GC。
public abstract class AbstractEndpoint<S> { public static interface Handler<S> { public enum SocketState { OPEN, CLOSED, LONG, ASYNC_END, SENDFILE, UPGRADING, UPGRADED, SUSPENDED } public SocketState process(SocketWrapperBase<S> socket, SocketEvent status); public Object getGlobal(); public Set<S> getOpenSockets(); public void release(SocketWrapperBase<S> socketWrapper); public void pause(); public void recycle(); } protected Handler<S> handler = null; public Handler<S> getHandler() { return handler; } protected SynchronizedStack<SocketProcessorBase<S>> processorCache; protected abstract SocketProcessorBase<S> createSocketProcessor( SocketWrapperBase<S> socketWrapper, SocketEvent event); private Executor executor = Executors.newFixedThreadPool(5);; public Executor getExecutor() { return executor; } public boolean processSocket(SocketWrapperBase<S> socketWrapper, SocketEvent event, boolean dispatch) { try { if (socketWrapper == null) { return false; } SocketProcessorBase<S> sc = processorCache.pop(); if (sc == null) { sc = createSocketProcessor(socketWrapper, event); } else { sc.reset(socketWrapper, event); } Executor executor = getExecutor(); if (dispatch && executor != null) { executor.execute(sc); } else { sc.run(); } } catch (Throwable ree) { return false; } return true; } }
第四步:在NioPoint中实现createSocketProcessor抽象方法
protected SocketProcessorBase<NioChannel> createSocketProcessor( SocketWrapperBase<NioChannel> socketWrapper, SocketEvent event) { return new SocketProcessor(socketWrapper, event); }
第五步:新建ConnectionHandler类并实现Handelr接口
public class ConnectionHandler <S> implements AbstractEndpoint.Handler<S>{ @Override public SocketState process(SocketWrapperBase<S> socket, SocketEvent status) { Processor processor = createProcessor(); processor.service(socket); return null; } @Override public Object getGlobal() { return null; } @Override public Set<S> getOpenSockets() { return null; } @Override public void release(SocketWrapperBase<S> socketWrapper) { } @Override public void pause() { } @Override public void recycle() { } protected Processor createProcessor() { Http11Processor processor = new Http11Processor(); return processor; }
第六步:在NioEndpoint类中的startInternal方法中引用ConnectionHandler。初始化processorCache大小
public void startInternal() { try { processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE, 500); handler = new ConnectionHandler<>(); pollers = new Poller[pollerThreadCount]; for (int i=0; i<pollers.length; i++) { pollers[i] = new Poller(); Thread pollerThread = new Thread(pollers[i]); // 设置线程优先级 pollerThread.setPriority(5); // 设置守护线程 pollerThread.setDaemon(true); pollerThread.start(); } Runnable acceptor = new Acceptor(); new Thread(acceptor).start(); } catch (Exception e) { } }
第七步:新建SynchronizedStack。
public class SynchronizedStack<T> { public static final int DEFAULT_SIZE = 128; private static final int DEFAULT_LIMIT = -1; private int size; private final int limit; private int index = -1; private Object[] stack; public SynchronizedStack() { this(DEFAULT_SIZE, DEFAULT_LIMIT); } public SynchronizedStack(int size, int limit) { if (limit > -1 && size > limit) { this.size = limit; } else { this.size = size; } this.limit = limit; stack = new Object[size]; } public synchronized boolean push(T obj) { index++; if (index == size) { if (limit == -1 || size < limit) { expand(); } else { index--; return false; } } stack[index] = obj; return true; } public synchronized T pop() { if (index == -1) { return null; } T result = (T) stack[index]; stack[index--] = null; return result; } public synchronized void clear() { if (index > -1) { for (int i = 0; i < index + 1; i++) { stack[i] = null; } } index = -1; } private void expand() { int newSize = size * 2; if (limit != -1 && newSize > limit) { newSize = limit; } Object[] newStack = new Object[newSize]; System.arraycopy(stack, 0, newStack, 0, size); stack = newStack; size = newSize; } }
第八步:新建Processor接口
public interface Processor { public AbstractEndpoint.Handler.SocketState service(SocketWrapperBase<?> socketWrapper); }
第九步:新建Http11Processor类并实现Processor接口
public class Http11Processor implements Processor { protected final Http11InputBuffer inputBuffer; public Http11Processor() { inputBuffer = new Http11InputBuffer(); } /** * 在这里解析请求行和请求头 */ @Override public AbstractEndpoint.Handler.SocketState service(SocketWrapperBase<?> socketWrapper) { setSocketWrapper(socketWrapper); try { inputBuffer.parseRequestLine(); inputBuffer.parseHeaders(); } catch (Exception e) { // ignore } return null; } protected final void setSocketWrapper(SocketWrapperBase<?> socketWrapper) { inputBuffer.init(socketWrapper); } }
第十步:在NioChannel中新建setPoller方法
public void setPoller(NioEndpoint.Poller poller) { this.poller = poller; }
第十一步:在Poller类中的register方法中调用setPoller方法
![](/i/l/?n=23&i=blog/1140628/202212/1140628-20221210142808054-2079539430.png)
运行结果
下面是本章程序中处理一次请求的时序图
结束!!!
标签:null,return,请求,void,socketWrapper,public,源码,接收,event From: https://www.cnblogs.com/yishi-san/p/16971551.html