नेट्टी उच्च प्रदर्शन NIO सर्वर

प्रस्तावना

नमस्ते मैं CIS में सबसे बड़े Minecraft सर्वर का मुख्य विकासक हूं (मैं विज्ञापन नहीं दूंगा कि इसकी जरूरत किसे है, वे जानते हैं)। लगभग एक वर्ष से अब हम अपने सर्वर कार्यान्वयन को लिख रहे हैं, जिसे 40 से अधिक लोगों के लिए डिज़ाइन किया गया है (हम कम से कम 500 का आंकड़ा देखना चाहते हैं)। अब तक, सब कुछ सफल था, लेकिन हाल ही में सिस्टम ने इस तथ्य पर आराम करना शुरू कर दिया कि सबसे सफल नेटवर्क कार्यान्वयन (इनपुट के लिए 1 स्ट्रीम, आउटपुट के लिए 1 + प्रसंस्करण के लिए 1) नहीं होने के कारण, 300 खिलाड़ियों के साथ ऑनलाइन, 980 से अधिक प्रवाह काम (सिस्टम) , जो डिफ़ॉल्ट io जावा के प्रदर्शन के साथ संयुक्त है, प्रदर्शन में भारी गिरावट देता है, और यहां तक ​​कि 100 खिलाड़ियों के साथ, सर्वर मुख्य रूप से नेटवर्क से लिखने / पढ़ने / करने में लगा हुआ है।

इसलिए मैंने NIO में जाने का फैसला किया। नेटी लाइब्रेरी गलती से हाथों में गिर गई, जिसकी संरचना इसे पहले से ही काम कर रहे समाधान में एकीकृत करने के लिए एकदम सही लग रही थी। दुर्भाग्य से, न केवल रूसी में, बल्कि अंग्रेजी में भी कुछ नेटी मैनुअल हैं, इसलिए मुझे सबसे अच्छा तरीका खोजने के लिए लाइब्रेरी कोड में बहुत प्रयोग करना और चढ़ाई करना पड़ा।

यहां मैं नेट्टी के माध्यम से नेटवर्क के साथ काम करने के सर्वर भाग को चित्रित करने की कोशिश करूंगा, शायद यह किसी के लिए उपयोगी होगा।

सर्वर निर्माण

ExecutorService bossExec = new OrderedMemoryAwareThreadPoolExecutor(1, 400000000, 2000000000, 60, TimeUnit.SECONDS); ExecutorService ioExec = new OrderedMemoryAwareThreadPoolExecutor(4 /*    */, 400000000, 2000000000, 60, TimeUnit.SECONDS); ServerBootstrap networkServer = new ServerBootstrap(new NioServerSocketChannelFactory(bossExec, ioExec, 4 /*       */)); networkServer.setOption("backlog", 500); networkServer.setOption("connectTimeoutMillis", 10000); networkServer.setPipelineFactory(new ServerPipelineFactory()); Channel channel = networkServer.bind(new InetSocketAddress(address, port)); 

फ्रेंच सहयोगियों के अनुभव के अनुसार, Netty कार्यों को करने के लिए OrderedMemoryAwareThreadPoolExecutor द्वारा उपयोग किया जाता है, वे सबसे प्रभावी हैं। आप अन्य Executors का उपयोग कर सकते हैं, जैसे Executors.newFixedThreadPool (n) । किसी भी स्थिति में Executors.newCachedThreadPool () का उपयोग न करें, यह अनुचित रूप से कई थ्रेड बनाता है और नेट्टी से लगभग कोई लाभ नहीं है। यह 4 से अधिक वर्कफ़्लो का उपयोग करने के लिए कोई मतलब नहीं है, जैसा कि वे अधिक भार का सामना करने से अधिक हैं (4 थ्रेड्स पर ज़ेबिया-फ्रांस के प्रोग्रामर 100,000 से अधिक एक साथ कनेक्शन खींचते हैं)। प्रत्येक सुनने वाले बंदरगाह के लिए बॉस धाराएँ एक होनी चाहिए। चैनल जो बाइंड फ़ंक्शन को लौटाता है, साथ ही ServerBootsrap को भी सहेजा जाना चाहिए ताकि आप सर्वर को बाद में रोक सकें।

PipelineFactory

जिस तरह से क्लाइंट के कनेक्शन और पैकेज को प्रोसेस किया जाएगा, वह पाइपलाइनलाइन द्वारा निर्धारित किया जाता है, जो क्लाइंट के साथ एक चैनल खोलने पर, इसके लिए एक पाइपलाइन बनाता है, जिसमें चैनल पर होने वाले इवेंट हैंडलर को परिभाषित किया जाता है। हमारे मामले में, यह ServerPipelineFactory है :

 public class ServerPipelineFactory implements ChannelPipelineFactory { @Override public ChannelPipeline getPipeline() throws Exception { PacketFrameDecoder decoder = new PacketFrameDecoder(); PacketFrameEncoder encoder = new PacketFrameEncoder(); return Channels.pipeline(decoder, encoder, new PlayerHandler(decoder, encoder)); } } 

इस कोड में, PacketFrameDecoder , PacketFrameEncoder, और PlayerHandler ईवेंट हैंडलर हैं जिन्हें हम परिभाषित करते हैं। Channels.pipeline () फ़ंक्शन हैंडलर के साथ एक नई पाइपलाइन बनाता है जो इसे पारित कर दिया गया है। सावधान रहें: घटनाओं को उस क्रम में संचालकों द्वारा संसाधित किया जाता है जिसमें आप पाइपलाइन फ़ंक्शन से गुजरे थे!

प्रोटोकॉल

मैं प्रोटोकॉल का थोड़ा वर्णन करूंगा ताकि बाद में यह स्पष्ट हो।

पैकेट क्लास का विस्तार करने वाले क्लास ऑब्जेक्ट का उपयोग करके डेटा का आदान-प्रदान किया जाता है, जिसमें दो फ़ंक्शन परिभाषित होते हैं, प्राप्त करें (ChannelBuffer इनपुट) और भेजें (ChannelBuffer आउटपुट) । तदनुसार, पहला फ़ंक्शन चैनल से आवश्यक डेटा पढ़ता है, दूसरा चैनल को पैकेट डेटा लिखता है।

 public abstract class Packet { public static Packet read(ChannelBuffer buffer) throws IOException { int id = buffer.readUnsignedShort(); //  ID  ,  ,     Packet packet = getPacket(id); //      ID if(packet == null) throw new IOException("Bad packet ID: " + id); //         ,   packet.get(buffer); //       return packet; } public statuc Packet write(Packet packet, ChannelBuffer buffer) { buffer.writeChar(packet.getId()); //  ID  packet.send(buffer); //    } // ,       public abstract void get(ChannelBuffer buffer); public abstract void send(ChannelBuffer buffer); } 

स्पष्टता के लिए संकुल की एक जोड़ी का एक उदाहरण:

 // ,       public class Packet1Login extends Packet { public String login; public void get(ChannelBuffer buffer) { int length = buffer.readShort(); StringBuilder builder = new StringBuilder(); for(int i = 0; i < length ++i) builder.append(buffer.readChar()); login = builder.toString(); } public void send(ChannelBuffer buffer) { //   , ..      } } // ,       ,      public class Packet255KickDisconnect extends Packet { public String reason; public void get(ChannelBuffer buffer) { int length = buffer.readShort(); StringBuilder builder = new StringBuilder(); for(int i = 0; i < length ++i) builder.append(buffer.readChar()); reason = builder.toString(); } public void send(ChannelBuffer buffer) { buffer.writeShort(reason.length()); for(int i = 0; i < reason.length(); ++i) { buffer.writeChar(reason.getCharAt(i)); } } } 

ChannelBuffer DataInputStream और DataOutputStream के समान है और सभी एक में लुढ़के हैं। अधिकांश फ़ंक्शन, यदि समान नहीं हैं, तो बहुत समान हैं। ध्यान दें कि मुझे जांचने की परवाह नहीं है कि क्या पढ़ने के लिए बफर में पर्याप्त बाइट्स हैं, जैसे कि मैं अवरुद्ध आईओ के साथ काम कर रहा था। इस पर और बाद में ...

ग्राहक सेवा

क्लाइंट के साथ काम मुख्य रूप से प्लेयरहैंडलर वर्ग द्वारा निर्धारित किया जाता है:

 public class PlayerHandler extends SimpleChannelUpstreamHandler { private PlayerWorkerThread worker; @Override public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { //     .    Worker  — ,      . //      ( e.getChannel()),        worker = new PlayerWorkerThread(this, e.getChannel()); } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { //   .   ,   ,    ,     .    ,     ,   ,       ,    . worker.disconnectedFromChannel(); } @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) { //     Packet'  ,       worker.      . if(e.getChannel().isOpen()) worker.acceptPacket((Packet) e.getMessage()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) { //    .  ,  . Server.logger.log(Level.WARNING, "Exception from downstream", e.getCause()); ctx.getChannel().close(); } } 

कार्यकर्ता चैनल को केवल खिलाड़ी को डेटा भेज सकता है। राइट (पैकेट) फ़ंक्शन, जहां चैनल खिलाड़ी का चैनल है, जो कनेक्ट होने पर उसे प्रेषित किया जाता है, और पैकेट पैकेट वर्ग का उद्देश्य है। एन्कोडर पैकेट एन्कोडिंग के लिए जिम्मेदार होगा।

विकोडक और एनकोडर

दरअसल, सिस्टम का सबसे महत्वपूर्ण हिस्सा यह है कि वे उपयोगकर्ता स्ट्रीम से पैकेट पैकेट बनाने और उसी पैकेट को स्ट्रीम में भेजने के लिए जिम्मेदार हैं।

एनकोडर बहुत सरल है, यह खिलाड़ी को पैकेट भेजता है:

 public class PacketFrameEncoder extends OneToOneEncoder { @Override protected Object encode(ChannelHandlerContext channelhandlercontext, Channel channel, Object obj) throws Exception { if(!(obj instanceof Packet)) return obj; //    ,      Packet p = (Packet) obj; ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(); //          .      ,       — ChannelBuffers     ,    . Packet.write(p, buffer); //     return buffer; //  ,       } } 

विकोडक पहले से ही बहुत अधिक जटिल है। तथ्य यह है कि क्लाइंट से आए बफर में, पूरे पैकेट को पढ़ने के लिए पर्याप्त बाइट्स नहीं हो सकते हैं। इस मामले में, रिप्लेइंगडेकोडर वर्ग हमारी मदद करेगा। हमें बस इसके डिकोड फंक्शन को लागू करने और किसी भी चीज की चिंता किए बिना इसमें स्ट्रीम से डेटा पढ़ने की जरूरत है:

 public class PacketFrameDecoder extends ReplayingDecoder<VoidEnum> { @Override public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { ctx.sendUpstream(e); } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { ctx.sendUpstream(e); } @Override protected Object decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer buffer, VoidEnum e) throws Exception { return Packet.read(buffer); } } 

सवाल यह है कि यह कैसे काम करता है? बहुत सरलता से, डिकोड फ़ंक्शन को कॉल करने से पहले, डिकोडर वर्तमान रीड इंडेक्स को चिह्नित करता है, अगर बफर से पढ़ते समय पर्याप्त डेटा नहीं है, तो एक अपवाद फेंक दिया जाएगा। इस स्थिति में, बफर अपनी प्रारंभिक स्थिति में वापस आ जाएगा और उपयोगकर्ता से अधिक डेटा प्राप्त होने पर डिकोड दोहराया जाएगा। एक सफल रीड (नॉन-नल लौटा) के मामले में, डिकोडर फिर से डीकोड फ़ंक्शन को कॉल करने की कोशिश करेगा, पहले से ही बफर में शेष डेटा पर, अगर इसमें कम से कम एक और बाइट है।

क्या यह सब धीरे-धीरे काम करता है अगर यह एक अपवाद फेंकता है? उदाहरण के लिए, बफर में डेटा की मात्रा की जांच करना और मूल्यांकन करना कि वे पैकेट पढ़ने के लिए पर्याप्त हैं या नहीं। लेकिन यह एक कैश्ड अपवाद का उपयोग करता है, इसलिए स्टैकट्रेस को भरने और यहां तक ​​कि एक नया अपवाद ऑब्जेक्ट बनाने में कोई समय खर्च नहीं किया जाता है। आप यहाँ ReplayingDecoder की सुविधाओं को बढ़ाने वाले कुछ अन्य प्रदर्शनों के बारे में अधिक पढ़ सकते हैं ।

आप फ़्रेमडेकोडर के साथ भी प्रयोग कर सकते हैं, उदाहरण के लिए, आप किसी पैकेट के आकार को उसकी आईडी द्वारा पूर्व-निर्धारित कर सकते हैं।

ऐसा लगता है कि यह सब है

परिणाम बेहतरीन थे। सबसे पहले, सर्वर अब एक हजार थ्रेड्स के साथ स्ट्रीम करता है - 4 नेट्टी स्ट्रीम + 4 डेटा प्रोसेसिंग थ्रेड्स 250+ क्लाइंट्स के साथ एक उत्कृष्ट कार्य करते हैं (परीक्षण जारी है)। दूसरे, प्रोसेसर लोड बहुत छोटा हो गया है और कनेक्शन की संख्या से रैखिक रूप से बढ़ना बंद हो गया है। तीसरे, कुछ मामलों में प्रतिक्रिया समय कम हो गया है।

मुझे आशा है कि किसी को यह मददगार लगेगा। मैंने जितना संभव हो उतना महत्वपूर्ण डेटा स्थानांतरित करने की कोशिश की, मैं बहुत दूर जा सकता था। आखिरकार, कई उदाहरण नहीं हैं? अपने जवाब पूछें और कड़ाई से न्याय न करें - मैं पहली बार हैबर में लिखता हूं।

पोस्टस्क्रिप्ट: कुछ और उपयोगी चीजें

नेट्टी में कुछ अन्य रोचक विशेषताएं हैं जो विशेष उल्लेख के योग्य हैं:

सबसे पहले, सर्वर बंद:

 ChannelFuture future = channel.close(); future.awaitUninterruptibly(); 

जहां चैनल वह चैनल है जहां बाइंड फंक्शन शुरुआत में लौटा था। जब तक चैनल बंद नहीं होगा और कोड का निष्पादन जारी रहेगा, तब तक future.awaitUninterruptibly () प्रतीक्षा करेगा ।

मुख्य विशेषताएं: ChannelFuture। जब हम Channel.write (पैकेट) फ़ंक्शन का उपयोग करके किसी चैनल को पैकेट भेजते हैं, तो वह ChannelFuture पर लौटता है - यह एक विशेष ऑब्जेक्ट है जो प्रदर्शन की जा रही स्थिति की निगरानी करता है। इसके माध्यम से, आप जांच सकते हैं कि कार्रवाई पूरी हो गई है या नहीं।

उदाहरण के लिए, हम एक ग्राहक को एक डिस्कनेक्ट पैकेट भेजना चाहते हैं और उसके पीछे चैनल बंद कर सकते हैं। अगर हम करते हैं

 channel.write(new Packet255KickDisconnect("!")); channel.close(); 

फिर 99% की संभावना के साथ, हमें ChannelClosedException मिलेगी और पैकेट क्लाइंट तक नहीं पहुंचेगा। लेकिन आप ऐसा कर सकते हैं:

 ChannelFuture future = channel.write(new Packet255KickDisconnect("!")); try { future.await(10000); //    10 ,    } catch(InterruptedException e) {} channel.close(); 

सब कुछ सुपर होगा, सिवाय इसके कि यह निष्पादन के धागे को ब्लॉक कर सकता है जब तक कि पैकेट उपयोगकर्ता को नहीं भेजा जाता है। इसलिए, आप ChannelFuture पर एक श्रोता को लटका सकते हैं - एक ऐसी वस्तु जिसे अधिसूचित किया जाएगा कि घटना पूरी हो गई है और कोई कार्रवाई करेगा। कनेक्शन को बंद करने के लिए एक तैयार श्रोता ChannelFutureListener.CLOSE है । उपयोग उदाहरण:

 ChannelFuture future = channel.write(new Packet255KickDisconnect("!")); furute.addListener(ChannelFutureListener.CLOSE); 

प्रभाव एक ही है, कोई ताले नहीं हैं। यह पता लगाना मुश्किल नहीं है कि अपना श्रोता कैसे बनाया जाए - केवल एक ही कार्य है। किसी भी तैयार वर्ग को खोलें, यहां मैं एक उदाहरण नहीं दूंगा।

अधिक महत्वपूर्ण जानकारी

जैसा कि टिप्पणियों में सही ढंग से नोट किया गया था, यह चेतावनी दी जानी चाहिए कि अवरुद्ध संचालन या हैंडलर में इंतजार नहीं करना बेहतर है (हैंडलर जो पाइप लाइन पर लटका हुआ है)। अन्यथा, आप स्थायी रूप से प्रसंस्करण के प्रवाह को खोने या अन्य ग्राहकों की घटनाओं के प्रसंस्करण को धीमा करने का जोखिम चलाते हैं।

साथ ही, हैंडलर में, किसी भी स्थिति में "भविष्य की प्रतीक्षा" नहीं कर सकता, अर्थात। execute .wait () या .awaitUninterruptibly () किसी भी ChannelFuture पर। सबसे पहले, आप सफल नहीं हुए हैं, आप उन्हें हैंडलर से नहीं बुला सकते हैं - सिस्टम आपको ऐसी मूर्खता नहीं करने देगा और एक अपवाद को फेंक देगा। दूसरे, यदि यह नहीं था, तो आपका प्रवाह फिर से सेवा के बिना अन्य ग्राहकों को छोड़कर मर सकता है।

सामान्य तौर पर, ChannelHandlers में किए गए सभी कार्य यथासंभव सरल और गैर-अवरुद्ध होने चाहिए। किसी भी मामले में डेटा को सीधे उन में संसाधित न करें - पैकेट को कतार में रखें और उन्हें दूसरे धागे में संसाधित करें।