Shulou(Shulou.com)06/01 Report--

This article introduces you how the principle of MINA is, the content is very detailed, interested friends can refer to, hope to be helpful to you.

1. Establish a connection with the server through SocketConnector

two。 After the link is established, the read and write of the IProcessor O is handed over to the Imax O Processor thread, which is multithreaded.

3. The data read by I Processor go through all the configured IoFilter,IoFilter in IoFilterChain for message filtering and format conversion. At this level, some custom protocols can be developed.

4. Finally, IoFilter gives the data to Handler for business processing, and completes the whole reading process.

5. The writing process is also similar, except that the data is written out through IoSession.write, and then the Handler carries out the business processing of writing. After the processing is completed, it is handed over to IoFilterChain for message filtering and protocol conversion. Finally, the data is written out to the socket channel through I _ Processor.

IoFilterChain acts as a message filter chain

1. When reading, it is the process from low-level protocol to high-level protocol, generally speaking, from byte bytes to business objects.

two。 Writing is usually a process from a business object to a byte byte

IoSession runs through the whole communication process.

The whole process can be represented by a diagram.

Message arrows are all initiated by NioProcessor-N threads, which are also executed in NioProcessor-N threads by default.

Connector: as a connection client, SocketConector is used to establish a connection with the server. If the connection is successful, an IoProcessor Thread is created (no more than the specified processorCount). The Thread is managed by the specified thread pool. IoProcessor uses the NIO framework to process the IO and create the IoSession at the same time. The connection is established through the SocketChannel of Nio.

NioSocketConnector connector = new NioSocketConnector (processorCount)

ConnectFuture future = connector.connect (new InetSocketAddress (HOSTNAME, PORT)); set up an Icano channel

Acceptor: as the connection recipient on the server side, SocketAcceptor is used to listen on the port and establish a connection with the client. After the connection is established, all the IoProcessor O operations are handed over to IoProcessor for processing.

IoAcceptor acceptor = new NioSocketAcceptor ()

Acceptor.bind (new InetSocketAddress (PORT))

Protocol: use IoFilter to decode and encode messages. For example, the following code converts java objects into byte strings through MyProtocolEncoder, and restores byte strings into java objects through MyProtocalDecoder.

Connector.getFilterChain () .addLast ("codec";, new ProtocolCodecFilter (new MyProtocalFactory ())

.

Public class MyProtocalFactory implements ProtocolCodecFactory {

ProtocolEncoderAdapter encoder = new MyProtocolEncoder ()

ProtocolDecoder decoder = new MyProtocalDecoder ()

Public ProtocolDecoder getDecoder (IoSession session) throws Exception {

Return decoder

}

Public ProtocolEncoder getEncoder (IoSession session) throws Exception {

Return encoder

}

}

.

Public class MyProtocalDecoder extends ProtocolDecoderAdapter {

Public void decode (IoSession session, IoBuffer in, ProtocolDecoderOutput out)

Throws Exception {

Int id = in.getInt ()

Int len = in.getInt ()

Byte [] dst = new byte [len]

In.get (dst)

String name = new String (dst, "GBK")

Item item = new Item ()

Item.setId (id)

Item.setName (name)

Out.write (item)

}

}

.

Public class MyProtocolEncoder extends ProtocolEncoderAdapter {

Public void encode (IoSession session, Object message

ProtocolEncoderOutput out) throws Exception {

Item item = (Item) message

Int byteLen = 8 + item.getName () .getBytes ("GBK") .length

IoBuffer buf = IoBuffer.allocate (byteLen)

Buf.putInt (item.getId ())

Buf.putInt (item.getName () .getBytes ("GBK") .length)

Buf.put (item.getName () .getBytes ("GBK";)

Buf.flip ()

Out.write (buf)

}

}

Handler: specific handling of events, including: sessionCreated, sessionOpened, sessionClosed, sessionIdle, exceptionCaught, messageReceived, messageSent.

Connector.setHandler (new MyHandler ()); MyHandler inherits the IoHandlerAdapter class or implements the IoHandler interface. The event is finally called by the IoProcessor thread.

Processor: the Nio O processor, allows multiple threads to read and write, and only needs to specify the number of threads in the development process. Processor uses the Nio framework to continue writing to the Nio O, and Processor contains a reference to the Nio's Selector. This is also the advantage of mina. If you write it directly in Nio, you need to write your own code to implement functions similar to Processor. Because Processor O Processor processes reading and writing asynchronously, we sometimes need to identify messages of the same task, for example, a task includes sending messages, receiving messages, and feedback messages. Then we need to include an id in the message header that can identify the same task when formulating the message format.

Setting the number of SocketConnector O Porcessor threads: if it is SocketConnector, you can specify it in the constructor, such as new SocketConnector (processorCount, Executors.newCachedThreadPool ()); if it is SocketAcceptor, it is the same: SocketAcceptor acceptor = new SocketAcceptor (ProcessorCount, Executors.newCachedThreadPool ())

ProcessorCount is the maximum number of portor threads, which can be tuned through performance tests, and the default value is the number of cpu cores + 1 (Runtime.getRuntime (). AvailableProcessors () + 1).

The strange thing is that each IoProcessor will establish a connection with itself locally when it is created.

IoSession: IoSession is used to maintain the context of IoService. An IoService establishes an IoSession (a connection and a session) after the establishment of a Connect, and the life cycle of the IoSession is from the establishment of the Connection to the disconnection.

IoSession does two things:

1. Through IoSession, you can get all relevant configuration objects for IoService (holding references to IoService,Processor pools, SocketChannel,SessionConfig, and IoService.IoHandler)

two。 IoSession.write is the entry point for data writing.

About Thread

ThreadModel version 1.x of mina also has a threading mode option that is not available after 2.x.

Version 1.x specifies thread mode

SocketConnectorConfig cfg = new SocketConnectorConfig ()

Cfg.setThreadModel (ThreadModel.MANUAL)

MINA has three types of worker threads

Acceptor, Connector, Ideband O processor threads

Acceptor Thread: generally, as the receiving thread of the server-side link, it implements the interface IoService. The number of threads is the number of SocketAcceptor created.

Connector Thread: generally set up a link thread as a request from the client, implement the interface IoService, and maintain a link to the server-side Acceptor. The number of threads is the number of SocketConnector created.

Both SocketAcceptor and SocketConnector of Mina inherit BaseIoService and are two different implementations of IoService.

Processor Thread O processor Thread: the thread that is actually processed by CPU O exists on the server and client side, and is used to handle read and write operations. The number of threads can be configured. The maximum number of threads is + 1 by default.

Server side: specify ProcessorCount when creating SocketAcceptor

SocketAcceptor acceptor = new SocketAcceptor (Runtime.getRuntime (). AvailableProcessors () + 1, Executors.newCachedThreadPool ())

Client: specify ProcessorCount when creating SocketConnector

SocketConnector connector = new SocketConnector (Runtime.getRuntime (). AvailableProcessors () + 1, Executors.newCachedThreadPool ())

SocketConnector O Processor Thread, which is attached to IoService, is similar to the above example SocketConnector connector = new SocketConnector (Runtime.getRuntime (). AvailableProcessors () + 1, Executors.newCachedThreadPool ()); it means that the thread SocketConnector allows the number of CPU+1 Processor Thread.

Although NioProcessor is multithreaded, only one thread is used for business processing when connecting to one (Processor thread uses only one thread NioProcessor-n for a client connection) if handler's business is time-consuming, it will cause NioProcessor thread congestion, and a second thread will be created when two clients connect at the same time (provided that the first NioProcessor is busy) The maximum number to create is specified by the Acceptor constructor. If: a client connection has a lot of communication with the server, and the cost of Handler O is small, but the business processing time of Handler is relatively long, then you need to adopt a separate thread mode and add an ExecutorFitler at the end of the server:

Acceptor.getFilterChain () .addLast ("threadPool", new ExecutorFilter (Executors.newCachedThreadPool ())

This ensures that the threads of processor and handler are separate, otherwise: the client sends three messages, and the server processes each message for about 10s, then the three messages are processed serially, and the subsequent messages will be blocked when the first message is processed, and the client has the same problem.

Client-side Porcessor congestion test:

1. The following code sends 5 consecutive messages (item) after the connection is established

ConnectFuture future = connector.connect (new InetSocketAddress (HOSTNAME, PORT))

Future.awaitUninterruptibly ()

Session = future.getSession ()

Item item = new Item ()

Item.setId (12345)

Item.setName ("hi")

Session.write (item)

Session.write (item)

Session.write (item)

Session.write (item)

Session.write (item)

two。 The delay processing is carried out in the messageSent method of handle, and the delay is 3 seconds.

Public void messageSent (IoSession session, Object message) throws Exception {

Thread.sleep (3000)

System.out.println (message)

}

3. Test result

Five messages are sent serially and are all processed by the same IoPorcessor thread

Session.write (item)

Session.write (item)

Session.write (item)

Session.write (item)

Session.write (item)

The server receives a message every 3 seconds. Because the call is triggered by IoProcessor, and a connector uses only one IoProcessor thread

4. Add ExecutorFilter,ExecutorFilter to ensure that it is a separate thread when dealing with handler

Connector.getFilterChain () .addLast ("threadPool", new ExecutorFilter (Executors.newCachedThreadPool ())

5. Test result

Four session.wirte became parallel processing, and the server received five messages at the same time.

About how the principle of MINA is shared here, I hope that the above content can be of some help to you, can learn more knowledge. If you think the article is good, you can share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.