小文件系统的请求异步化高并发性能优化
222_分布式图片存储系统中的高性能指的到底是什么?
重构系统架构,来实现一个高性能。然后就要做非常完善的一个测试,最后对这个系统做一个总结,说说后续我们还要做一些什么东西。另外,我还要给大家留一些作业,相当于是让大家课后自己去做的,就不是完全拷贝我的代码
高并发
前面已经通过Reactor模式实现了
高性能主要是两块
第一块:客户端现在是短连接,每次发送请求,都需要建立连接,然后断开连接。站在客户端的角度而言,发现每执行一次文件上传和下载的操作,速度都很慢
第二块:文件上传,需要多副本上传。一般来说,针对kafka,多副本的时候默认情况下只要写成功一个副本,就返回了。另外其他的副本的写都是异步慢慢来执行的,kafka采取的是副本pull数据的机制,只要在一个数据节点上写成功数据,别的数据节点会主从从这个写成功的数据节点上pull数据
Kafka,强调高性能,生产消息的行为都是尽快的可以完成
HDFS,不强调高性能,它主要针对的是几个GB的大文件上传到服务器上去,只要慢慢上传就可以了,速度慢点无所谓,只要能上传成功。所以,HDFS采用的是多个副本一定要依次上传成功,才可以说是本次文件上传成功了。所以,HDFS的上传速度肯定是很慢的,因为它们根本不强调文件上传过程的高性能。所以Kafka和HDFS的应用场景本身就不相同
高性能架构的重构
- 短连接 -> 长连接;
- 同步上传多副本 -> 写一个副本,其他副本在后台慢慢的异步复制和拉取
这样,文件上传和文件下载,性能至少会提升好几倍
223_回头审视一下客户端的短连接模式有哪些问题?
除了客户端有NioClient以外,数据节点也有NioClient,因为他在进行数据节点扩缩容时,需要从其他的数据节点拷贝副本过来写入本地,这个过程使用短连接也无所谓,因为这个过程都是后台慢慢执行的,但是当然最好也是重构成长连接模式
224_初步实现用于进行网络管理的NetworkManager组件
225_在NetworkManager中实现核心线程无限循环进行poll操作
NetworkManager
/*** 网络连接管理器*/
public class NetworkManager {// 正在连接中public static final Integer CONNECTING = 1;// 已经建立连接public static final Integer CONNECTED = 2;// 多路复用Selectorprivate Selector selector;// 所有的连接private Map<String, SocketChannel> connections;// 每个数据节点的连接状态private Map<String, Integer> connectState;// 等待建立连接的机器private ConcurrentLinkedQueue<Host> waitingConnectHosts;public NetworkManager() {try {this.selector = Selector.open();} catch (IOException e) {e.printStackTrace();}this.connections = new ConcurrentHashMap<String, SocketChannel>();this.connectState = new ConcurrentHashMap<String, Integer>();this.waitingConnectHosts = new ConcurrentLinkedQueue<Host>();new NetworkPollThread().start();}/*** 尝试连接到数据节点的端口上去*/public void maybeConnect(String hostname, Integer nioPort) throws Exception {synchronized(this) {if(!connectState.containsKey(hostname)) {connectState.put(hostname, CONNECTING);waitingConnectHosts.offer(new Host(hostname, nioPort)); }while(connectState.get(hostname).equals(CONNECTING)) {wait(100);}}}/*** 尝试把排队中的机器发起连接的请求*/private void tryConnect() {try {Host host = null;SocketChannel channel = null;while((host = waitingConnectHosts.poll()) != null) {channel = SocketChannel.open(); channel.configureBlocking(false); channel.connect(new InetSocketAddress(host.hostname, host.nioPort)); channel.register(selector, SelectionKey.OP_CONNECT); }} catch (Exception e) {e.printStackTrace();}}// 网络连接的核心线程class NetworkPollThread extends Thread {@Overridepublic void run() {while(true) {tryConnect();}}}// 代表了一台机器class Host {String hostname;Integer nioPort;public Host(String hostname, Integer nioPort) {this.hostname = hostname;this.nioPort = nioPort;}}}
226_在无限循环的poll方法中完成网络连接的建立
public class NetworkManager {// 正在连接中public static final Integer CONNECTING = 1;// 已经建立连接public static final Integer CONNECTED = 2;// 网络poll操作的超时时间public static final Long POLL_TIMEOUT = 500L; // 多路复用Selectorprivate Selector selector;// 所有的连接private Map<String, SocketChannel> connections;// 每个数据节点的连接状态private Map<String, Integer> connectState;// 等待建立连接的机器private ConcurrentLinkedQueue<Host> waitingConnectHosts;public NetworkManager() {try {this.selector = Selector.open();} catch (IOException e) {e.printStackTrace();}this.connections = new ConcurrentHashMap<String, SocketChannel>();this.connectState = new ConcurrentHashMap<String, Integer>();this.waitingConnectHosts = new ConcurrentLinkedQueue<Host>();new NetworkPollThread().start();}/*** 尝试连接到数据节点的端口上去*/public void maybeConnect(String hostname, Integer nioPort) throws Exception {synchronized(this) {if(!connectState.containsKey(hostname)) {connectState.put(hostname, CONNECTING);waitingConnectHosts.offer(new Host(hostname, nioPort)); }while(connectState.get(hostname).equals(CONNECTING)) {wait(100);}}}// 网络连接的核心线程class NetworkPollThread extends Thread {@Overridepublic void run() {while(true) {tryConnect();poll();}}/*** 尝试把排队中的机器发起连接的请求*/private void tryConnect() {try {Host host = null;SocketChannel channel = null;while((host = waitingConnectHosts.poll()) != null) {channel = SocketChannel.open(); channel.configureBlocking(false); channel.connect(new InetSocketAddress(host.hostname, host.nioPort)); channel.register(selector, SelectionKey.OP_CONNECT); }} catch (Exception e) {e.printStackTrace();}}/*** 尝试完成网络连接、请求发送、响应读取*/private void poll() {SocketChannel channel = null;try {int selectedKeys = selector.select(500); if(selectedKeys <= 0) {return;}Iterator<SelectionKey> keysIterator = selector.selectedKeys().iterator(); while(keysIterator.hasNext()){ SelectionKey key = (SelectionKey) keysIterator.next(); keysIterator.remove(); // 如果是网络连接操作if(key.isConnectable()){ channel = (SocketChannel) key.channel();if(channel.isConnectionPending()){ while(!channel.finishConnect()) {Thread.sleep(100); }} System.out.println("完成与服务端的连接的建立......"); InetSocketAddress remoteAddress = (InetSocketAddress)channel.getRemoteAddress();connectState.put(remoteAddress.getHostName(), CONNECTED);connections.put(remoteAddress.getHostName(), channel);}}} catch (Exception e) {e.printStackTrace();if(channel != null) {try {channel.close();} catch (IOException e1) {e1.printStackTrace();}}}}}// 代表了一台机器class Host {String hostname;Integer nioPort;public Host(String hostname, Integer nioPort) {this.hostname = hostname;this.nioPort = nioPort;}}}
227_客户端的核心业务方法对要发送的请求进行封装
228_将封装好的请求放入NetworkManager的请求队列中
229_如何实现异步发送请求以及同步等待响应两个接口
230_对每个数据节点获取一个请求缓存起来等待发送
231_在核心的poll方法中将每个机器暂存等待的请求发送出去
232_在核心的poll方法中对机器返回的响应进行读取
拿到响应
客户端将请求发出以后,就每隔100ms轮询一次,有没有响应结果返回回来
233_客户端建立连接的过程中异常了该如何返回响应?
234_客户端发送请求过程中异常了该如何返回响应?
/*** 发送请求*/private void sendRequest(SelectionKey key, SocketChannel channel) {InetSocketAddress remoteAddress = null;try {remoteAddress = (InetSocketAddress) channel.getRemoteAddress();String hostname = remoteAddress.getHostName();// 获取要发送到这台机器的请求的数据NetworkRequest request = toSendRequests.get(hostname);ByteBuffer buffer = request.getBuffer();// 将请求发送到对方机器上去channel.write(buffer);while (buffer.hasRemaining()) {channel.write(buffer);}System.out.println("本次请求发送完毕......");key.interestOps(SelectionKey.OP_READ);} catch (Exception e) {e.printStackTrace();// 发送失败,就取消关注OP_WRITE事件key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);if (remoteAddress != null) {String hostname = remoteAddress.getHostName();NetworkRequest request = toSendRequests.get(hostname);if (request.needResponse()) {NetworkResponse response = new NetworkResponse();response.setHostname(hostname);response.setRequestId(request.getId());// 请求发送失败,则客户端手动构造一个响应response.setError(true);finishedResponses.put(request.getId(), response);} else {toSendRequests.remove(hostname);}}}}
完整代码
NioClient
*** 客户端的一个NIOClient,负责跟数据节点进行网络通信*/
public class NioClient {private NetworkManager networkManager;public NioClient() {this.networkManager = new NetworkManager();}/*** 发送一个文件过去*/public Boolean sendFile(String hostname, int nioPort, byte[] file, String filename, long fileLength) { // 先根据hostname来检查一下,跟对方机器的连接是否建立好了// 没有建立好,那么就直接在此建立连接; 建立好连接后,就把连接给缓存起来,以备下次使用try {// 如果此时还没跟那个数据节点建立好连接if(!networkManager.maybeConnect(hostname, nioPort)) {return false;}NetworkRequest request = createSendFileRequest(hostname, nioPort, file, filename, fileLength);networkManager.sendRequest(request); NetworkResponse response = networkManager.waitResponse(request.getId());if(response.error()) {// 请求发送失败,客户端自己构造的response,并将response.error 设置为truereturn false;}ByteBuffer buffer = response.getBuffer();String responseStatus = new String(buffer.array(), 0, buffer.remaining());System.out.println("收到" + hostname + "的响应:" + responseStatus);return responseStatus.equals(NetworkResponse.RESPONSE_SUCCESS);} catch (Exception e) {e.printStackTrace(); }return false;}/*** 构建一个发送文件的网络请求*/private NetworkRequest createSendFileRequest(String hostname, Integer nioPort, byte[] file, String filename, long fileLength) {NetworkRequest request = new NetworkRequest();ByteBuffer buffer = ByteBuffer.allocate(NetworkRequest.REQUEST_TYPE + NetworkRequest.FILENAME_LENGTH + filename.getBytes().length + NetworkRequest.FILE_LENGTH + (int)fileLength); buffer.putInt(NetworkRequest.REQUEST_SEND_FILE); buffer.putInt(filename.getBytes().length); buffer.put(filename.getBytes()); buffer.putLong(fileLength); buffer.put(file);buffer.rewind(); request.setId(UUID.randomUUID().toString()); request.setHostname(hostname); request.setNioPort(nioPort); request.setBuffer(buffer); request.setNeedResponse(true); return request;}}
NetworkManager
/*** 网络连接管理器*/
public class NetworkManager {// 正在连接中public static final Integer CONNECTING = 1;// 已经建立连接public static final Integer CONNECTED = 2;// 断开连接public static final Integer DISCONNECTED = 3;// 响应状态:成功public static final Integer RESPONSE_SUCCESS = 1;// 响应状态:失败public static final Integer RESPONSE_FAILURE = 2;// 网络poll操作的超时时间public static final Long POLL_TIMEOUT = 500L;// 多路复用Selectorprivate Selector selector;// 所有的连接private Map<String, SelectionKey> connections;// 每个数据节点的连接状态private Map<String, Integer> connectState;// 等待建立连接的机器private final ConcurrentLinkedQueue<Host> waitingConnectHosts;// 排队等待发送的网络请求private Map<String, ConcurrentLinkedQueue<NetworkRequest>> waitingRequests;// 马上准备要发送的网络请求private Map<String, NetworkRequest> toSendRequests;// 已经完成请求的响应private Map<String, NetworkResponse> finishedResponses;public NetworkManager() {try {this.selector = Selector.open();} catch (IOException e) {e.printStackTrace();}this.connections = new ConcurrentHashMap<String, SelectionKey>();this.connectState = new ConcurrentHashMap<String, Integer>();this.waitingConnectHosts = new ConcurrentLinkedQueue<Host>();this.waitingRequests = new ConcurrentHashMap<String, ConcurrentLinkedQueue<NetworkRequest>>();this.toSendRequests = new ConcurrentHashMap<String, NetworkRequest>();this.finishedResponses = new ConcurrentHashMap<String, NetworkResponse>();new NetworkPollThread().start();}/*** 尝试连接到数据节点的端口上去*/public Boolean maybeConnect(String hostname, Integer nioPort) {synchronized (this) {if (!connectState.containsKey(hostname) ||connectState.get(hostname).equals(DISCONNECTED)) {connectState.put(hostname, CONNECTING);waitingConnectHosts.offer(new Host(hostname, nioPort));}while (connectState.get(hostname).equals(CONNECTING)) {try {wait(100);} catch (InterruptedException e) {e.printStackTrace();}}if (connectState.get(hostname).equals(DISCONNECTED)) {return false;}return true;}}/*** 发送网络请求** @param request*/public void sendRequest(NetworkRequest request) {ConcurrentLinkedQueue<NetworkRequest> requestQueue =waitingRequests.get(request.getHostname());requestQueue.offer(request);}/*** 等待指定请求的响应*/public NetworkResponse waitResponse(String requestId) throws Exception {NetworkResponse response = null;while ((response = finishedResponses.get(requestId)) == null) {Thread.sleep(100);}toSendRequests.remove(response.getHostname());finishedResponses.remove(requestId);return response;}// 网络连接的核心线程class NetworkPollThread extends Thread {@Overridepublic void run() {while (true) {tryConnect();prepareRequests();poll();}}/*** 尝试把排队中的机器发起连接的请求*/private void tryConnect() {Host host = null;SocketChannel channel = null;while ((host = waitingConnectHosts.poll()) != null) {try {channel = SocketChannel.open();channel.configureBlocking(false);channel.connect(new InetSocketAddress(host.hostname, host.nioPort));channel.register(selector, SelectionKey.OP_CONNECT);} catch (Exception e) {e.printStackTrace();connectState.put(host.hostname, DISCONNECTED);}}}/*** 准备好要发送的请求*/private void prepareRequests() {for (String hostname : waitingRequests.keySet()) {// 看一下这台机器当前是否还没有请求马上就要发送出去了ConcurrentLinkedQueue<NetworkRequest> requestQueue =waitingRequests.get(hostname);if (!requestQueue.isEmpty() && !toSendRequests.containsKey(hostname)) {// 对这台机器获取一个派对的请求出来NetworkRequest request = requestQueue.poll();// 将这个请求暂存起来,接下来 就可以等待发送出去toSendRequests.put(hostname, request);// 让这台机器对应的连接关注的事件为OP_WRITESelectionKey key = connections.get(hostname);key.interestOps(SelectionKey.OP_WRITE);}}}/*** 尝试完成网络连接、请求发送、响应读取*/private void poll() {SocketChannel channel = null;try {int selectedKeys = selector.select(500);if (selectedKeys <= 0) {return;}Iterator<SelectionKey> keysIterator = selector.selectedKeys().iterator();while (keysIterator.hasNext()) {SelectionKey key = (SelectionKey) keysIterator.next();keysIterator.remove();channel = (SocketChannel) key.channel();// 如果是网络连接操作if (key.isConnectable()) {// 建立连接finishConnect(key, channel);} else if (key.isWritable()) {// 发送请求sendRequest(key, channel);} else if (key.isReadable()) {// 接收响应readResponse(key, channel);}}} catch (Exception e) {e.printStackTrace();if (channel != null) {try {channel.close();} catch (IOException e1) {e1.printStackTrace();}}}}/*** 完成跟机器的连接*/private void finishConnect(SelectionKey key, SocketChannel channel) {InetSocketAddress remoteAddress = null;try {remoteAddress = (InetSocketAddress) channel.getRemoteAddress();if (channel.isConnectionPending()) {while (!channel.finishConnect()) {Thread.sleep(100);}}System.out.println("完成与服务端的连接的建立......");waitingRequests.put(remoteAddress.getHostName(),new ConcurrentLinkedQueue<NetworkRequest>());connections.put(remoteAddress.getHostName(), key);// 将连接状态置为:已连接connectState.put(remoteAddress.getHostName(), CONNECTED);} catch (Exception e) {e.printStackTrace();if (remoteAddress != null) {connectState.put(remoteAddress.getHostName(), DISCONNECTED);}}}/*** 发送请求*/private void sendRequest(SelectionKey key, SocketChannel channel) {InetSocketAddress remoteAddress = null;try {remoteAddress = (InetSocketAddress) channel.getRemoteAddress();String hostname = remoteAddress.getHostName();// 获取要发送到这台机器的请求的数据NetworkRequest request = toSendRequests.get(hostname);ByteBuffer buffer = request.getBuffer();// 将请求发送到对方机器上去channel.write(buffer);while (buffer.hasRemaining()) {channel.write(buffer);}System.out.println("本次请求发送完毕......");key.interestOps(SelectionKey.OP_READ);} catch (Exception e) {e.printStackTrace();// 发送失败,就取消关注OP_WRITE事件key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);if (remoteAddress != null) {String hostname = remoteAddress.getHostName();NetworkRequest request = toSendRequests.get(hostname);if (request.needResponse()) {NetworkResponse response = new NetworkResponse();response.setHostname(hostname);response.setRequestId(request.getId());// 请求发送失败,则客户端手动构造一个响应response.setError(true);finishedResponses.put(request.getId(), response);} else {toSendRequests.remove(hostname);}}}}/*** 读取响应信息*/private void readResponse(SelectionKey key, SocketChannel channel) throws Exception {InetSocketAddress remoteAddress = (InetSocketAddress) channel.getRemoteAddress();String hostname = remoteAddress.getHostName();NetworkRequest request = toSendRequests.get(hostname);NetworkResponse response = null;if (request.getRequestType().equals(NetworkRequest.REQUEST_SEND_FILE)) {response = readSendFileResponse(request.getId(), hostname, channel);}key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);// 如果发送请求时,明确表示需要返回值if (request.needResponse()) {finishedResponses.put(request.getId(), response);} else {toSendRequests.remove(hostname);}}/*** 读取上传文件的响应*/private NetworkResponse readSendFileResponse(String requestId,String hostname, SocketChannel channel) throws Exception {ByteBuffer buffer = ByteBuffer.allocate(1024);channel.read(buffer);buffer.flip();NetworkResponse response = new NetworkResponse();response.setRequestId(requestId);response.setHostname(hostname);response.setBuffer(buffer);response.setError(false);return response;}}// 代表了一台机器class Host {String hostname;Integer nioPort;public Host(String hostname, Integer nioPort) {this.hostname = hostname;this.nioPort = nioPort;}}}