VPF::Проблема загрузки процессора - Форум программистов

acin

Дата 13.12.2010, 21:44 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 2
Регистрация: 13.12.2010

Репутация: нет
Всего: нет

Есть эмулятор сервера. Эмулятор работает по принципу перехвата пакетов клиента и отправки их на сервер. Сервреная и клиентская часть эмулятора написанна на NIO. Принцип работы прост: Клиент конектится к эмулятору, эмулятор к серверу(в отдельной нити для каждого клиента). Так вот при коннекте 1го клиента эмулятор грузит процессор(4 ядра) на 25%, при коннекте 2го уже на 98%. В чем может быть проблема??

VictorTsaregorodtsev

Дата 13.12.2010, 22:04 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 274
Регистрация: 28.7.2006

Репутация: нет
Всего: 8

Нисколько не спец в Джаве, но сделал бы так:
1. В диспетчере задач для графика загрузки ЦП включил бы показ времени ядра и посмотрел, что именно грузит проц: либо программа, либо вызываемые ей системные функции. Если второе - то надо менять методы/компоненты работы с сетью (либо как-то попытаться переделать алгоритм).
2. Если проц грузится именно несистемной работой, то профилировщик в руки и вперед смотреть времязатратные куски. Куда в первую очередь смотреть (на сервер или на клиента) - определил бы с помощью диспетчера задач во время выполнения п.1 (если, конечно, клиент и сервер представляют собой 2 разные проги, а не разные потоки внутри одной проги).

acin

Дата 13.12.2010, 22:51 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 2
Регистрация: 13.12.2010

Репутация: нет
Всего: нет

Цитата(VictorTsaregorodtsev @ 13.12.2010, 22:04

)

сервер представляют собой 2 разные проги

Как раз таки просто разные потоки. Вот код программы. Это главный поток. Обрабатывает клиентов.

Код


import java.io.*;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.*;
import java.nio.channels.*;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.concurrent.*;


import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;
/**
 * Класс Server. Делает все в одной нити. Принимает пакеты от клиента и отправляет в новую нить(класс Client). 
 */
public class Server implements Runnable
{
    protected static final Logger LOG = Logger.getLogger(Server.class);
    private final Selector selector;
    private final int clientPORT = 2106;
    private int BUFFER_SIZE = 64 * 1024;
    private final ByteOrder BYTE_ORDER = ByteOrder.LITTLE_ENDIAN;
    //Мапы, где хранятся все SelectionKey и связанные с ним ByteBuffer для чтения/записи
    public static ConcurrentHashMap<SelectionKey, ByteBuffer> syncReadClientPacketMap = new ConcurrentHashMap<SelectionKey, ByteBuffer>();
    public static ConcurrentHashMap<SelectionKey, ByteBuffer> syncReadServerPacketMap = new ConcurrentHashMap<SelectionKey, ByteBuffer>();
    public static ConcurrentMap<SelectionKey, Integer> syncWriteClientMap = new ConcurrentHashMap<SelectionKey, Integer>();
    public static ConcurrentMap<SelectionKey, Integer> syncWriteServerMap = new ConcurrentHashMap<SelectionKey, Integer>();
    /**
     * Конструктор объекта сервера
     * @throws IOException
     */
    public Server() throws IOException
    {
        selector = initConnection();
        LOG.info("Initialize");
    }
    /**
     * главный цикл прослушивания/ожидания коннекта.
     */
    public void run()
    {
        while (true)
        {
            try 
            {
                if (selector.isOpen())
                {
                    selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    for (SelectionKey sk:keys)
                    {
                        if (!sk.isValid())
                        {
                            continue;
                        }
                        if (sk.isAcceptable())
                        {
                            ServerSocketChannel ssca = (ServerSocketChannel)sk.channel(); 
                            SocketChannel sc = ssca.accept();
                            sc.configureBlocking(false);
                            SelectionKey skr = sc.register(selector, SelectionKey.OP_WRITE);
                            ByteBuffer byteBuffer = ByteBuffer.allocateDirect(BUFFER_SIZE).order(BYTE_ORDER);
                            
                            syncReadClientPacketMap.put(skr, byteBuffer);// мапа где хранятся буфферы принятые от клиента
                            syncReadServerPacketMap.put(skr, byteBuffer);// мапа где хранятся буфферы принятые от сервера
                            syncWriteServerMap.put(skr, 0);// мапа запрещающая писать в канал если пакет от сервера еще не прочитан. Для класса Client.
                            syncWriteClientMap.put(skr, 0);// аналогично выше, только для для класса Server
                            // 0 нельзя писать(буффер не готов), 1 можно
                            // для синхронизации чтени/записи между потоками ни чего больше не придумал
                            
                            LOG.info("New client connection");
                            Client client = new Client(skr);// создаем новую нить и передаем ей ключ нашего клиента
                            Thread t = new Thread(client);
                            t.start();
                        }
                        else if (sk.isReadable())
                        {
                            ReadableByteChannel socketChannel = (ReadableByteChannel)sk.channel(); // хватаем канал коннекта
                            int read;
                            ByteBuffer byteBuffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
                            byteBuffer.clear(); // очищаем байт буфер
                            LOG.info("Read client packet");
                            try
                            {
                                read = socketChannel.read(byteBuffer); // пробуем заполнить буфер
                                LOG.info("Read from client: "+read+"byte's");
                            }
                            catch (IOException e) // коннект отпал...
                            { 
                                closeChannel(sk); // закрываем
                                break; // выходим с цикла
                            }
                            if (read == -1) // коннект отвалился в штатном режиме
                            {
                                closeChannel(sk); // закрываем
                                break;// выходим с цикла
                            }
                            else if(read > 0)// если что-то прочитали из сокета и записали в буфер...
                            {
                                if (byteBuffer.remaining() > 0)
                                {
                                    syncReadClientPacketMap.replace(sk, byteBuffer);
                                    syncWriteServerMap.replace(sk, 1);
                                    sk.interestOps(SelectionKey.OP_WRITE); 
                                    byteBuffer = null;
                                }
                            }
                        }
                        else if (sk.isWritable() && syncWriteClientMap.get(sk) != 0)
                        {
                            LOG.info("Write client packet");
                            ByteBuffer bb  = syncReadServerPacketMap.get(sk);// получаем байт-буффер, ассоциированный с ключем
                            WritableByteChannel s = (WritableByteChannel)sk.channel(); // выдергиваем канал
                            try 
                            {
                                
                                bb.flip();
                                int result = s.write(bb); //пробуем записать
                                LOG.info("Write to client: "+result+"byte's");
                                if (result == -1)
                                {
                                    closeChannel(sk); //закрываем коннект
                                }
                            }
                            catch (IOException e2)
                            {
                                closeChannel(sk); //закрываем
                            }
                            if (bb.position() == bb.limit())
                            { 
                                sk.interestOps(SelectionKey.OP_READ);
                                syncWriteClientMap.replace(sk, 0);
                                bb = null;
                            }
                        }
                    }
                    keys.clear(); // очищаем сет ключей.
                }
                else
                    break;
            }
            catch (IOException e)
            {
                e.printStackTrace();
            }
        }
    }


     /**
     * Метод закрывает канал сокета, снимает со списка активных ключей и удаляет из списка рассылки
     * @param sk - ключ, связанный с каналом
     * @throws IOException
     */
    private void closeChannel(SelectionKey sk) throws IOException
    {
        LOG.info("Close client connection");
        syncReadClientPacketMap.remove(sk); // удаляем из списка рассылки
        syncReadServerPacketMap.remove(sk); // удаляем из списка рассылки
        SocketChannel socketChannel = (SocketChannel)sk.channel();
        if (socketChannel.isConnected())
        {
            socketChannel.close(); // закрываем канал
        }
        sk.cancel(); // удаляем из списка селектора
    }
     /**
     * Инициализация коннекта
     * @return Selector
     * @throws IOException
     */
    private Selector initConnection () throws IOException
    {
        ServerSocketChannel ssc;
        ssc = ServerSocketChannel.open(); // создаем серверСокет канал
        ssc.configureBlocking(false); // отключаем режим блокирования в ожидании
        ssc.socket().bind(new InetSocketAddress(clientPORT)); // получаем обычный серверсокет, который биндиться на нужный порт
        Selector sel = Selector.open(); // создаем селектор прослушки
        ssc.register(sel, SelectionKey.OP_ACCEPT); // регистрируемся на селекторе на сервер-канал.
        return sel;
    }
  
    public static void main(String[] args) throws IOException
    {
        BasicConfigurator.configure(new ConsoleAppender(new PatternLayout("%d{yyyyMMdd-HH:mm:ss} %-10t %-5p %-20C{1} - %m%n")));
        Logger.getRootLogger().setLevel(Level.INFO);
        Server srv = new Server();
        new Thread(srv).start();
    }

}

Следующий код это 2ой поток. Получает пакеты от первого потока и отправляет на сервер.

Код


import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Set;
import java.util.Iterator;

import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;


public class Client implements Runnable
{
    protected static final Logger LOG = Logger.getLogger(Client.class);
        
    private final int BUFFER_SIZE = 64 * 1024;//размер буффера
    private final ByteOrder BYTE_ORDER = ByteOrder.LITTLE_ENDIAN;// порядок байт буффера
    
    private Selector selector;
    private SocketChannel channel;
    private SocketAddress address;
    private SelectionKey skr;
    
    private final Thread thread = new Thread(this);
    /**
     * Инициализируем клиента для сервера, проставляем полученный ключ от класса Server.
     */
    public Client(SelectionKey skr) throws IOException
    {
        this.skr = skr;
        setAddress(new InetSocketAddress("127.0.0.1", 2105));
    }
    /**
     * главный цикл.
     */
    public void run()
    {
        LOG.info("event loop running");
        try
        {
            while(true)
            {
                try
                {
                    selector = Selector.open();
                    channel = SocketChannel.open();
                    configureChannel(channel);
                    channel.connect(address);
                    channel.register(selector, SelectionKey.OP_CONNECT);
                    
                    while(!thread.isInterrupted() && channel.isOpen())
                    { 
                        if (selector.select() > 0)
                        {
                            SelectedKeys(selector.selectedKeys());
                        }
                    }
                }
                catch (Exception e)
                {
                    LOG.error("exception", e);
                }
                finally
                {
                    if (channel != null)
                        channel.close();
                    if (selector != null)
                        selector.close();
                    LOG.info("connection closed");
                    terminate();
                }
            }
        }
        catch (Exception e)
        {
            LOG.error("unrecoverable error", e);
        }    
    }
    /**
     * Коннект.
     * @param key - ключ, связанный с каналом
     * @throws IOException
     */
    private void connect(SelectionKey key) throws IOException
    {
        SocketChannel ch = (SocketChannel) key.channel();
        if (ch.finishConnect())
        {
            LOG.info("connected to " + address);
            key.interestOps(key.interestOps() ^ SelectionKey.OP_CONNECT);
            key.interestOps(key.interestOps() | SelectionKey.OP_READ);
        }
    }
    /**
     * Запись пакета полученного от класса Server.
     * @param key - ключ, связанный с каналом
     * @throws IOException
     */
    private void write(SelectionKey key) throws IOException
    {
        LOG.info("Write server packet");        
        WritableByteChannel writeChannel = (WritableByteChannel)key.channel(); // выдергиваем канал
        ByteBuffer writeBuf = Server.syncReadClientPacketMap.get(skr);// получаем байт-буффер, ассоциированный с ключем клиента
        try 
        {
          writeBuf.flip();
            int result = writeChannel.write(writeBuf); // пробуем записать
            LOG.info("Write to server: "+result+"byte's");
            if (result == -1) //коннект отвалился в штатном режиме
            {
                closeChannel(key); //закрываем коннект
            }
        }
        catch (IOException e2)
        {
            closeChannel(key); //закрываем
        }
        if (writeBuf.position() == writeBuf.limit())
        { 
            key.interestOps(SelectionKey.OP_READ);
            Server.syncWriteServerMap.replace(skr, 0);
            writeBuf = null;
        }
    }
    /**
     * Чтение пакета для отправки классу Server.
     * @param key - ключ, связанный с каналом
     * @throws IOException
     */
    private void read(SelectionKey key) throws IOException
    {
        LOG.info("Read server packet");
        ReadableByteChannel readChannel = (ReadableByteChannel)key.channel();
        ByteBuffer readBuf = ByteBuffer.allocateDirect(BUFFER_SIZE);
        readBuf.clear();
        int read = 0;
        try
        {
            read = readChannel.read(readBuf); // пробуем заполнить буфер
            LOG.info("Read from server: "+read+"byte's");
        }
        catch (IOException e) // коннект отпал...
        { 
           closeChannel(key); // закрываем
        }
        if (read == -1) // коннект отвалился в штатном режиме
        {
            closeChannel(key); // тоже закрываем
        }
        else if (read > 0)
        {
            key.interestOps(SelectionKey.OP_WRITE);
            Server.syncReadServerPacketMap.replace(skr, readBuf);// записываем полученный буффер в мапу
            Server.syncWriteClientMap.replace(skr, 1);// ставим ключ класса Server на возможность записать полученный буффер
            readBuf = null;
        }
    }
    /**
     * Цикл перебора ключей.
     * @param key - сэт ключей
     * @throws Exception
     */
    private void SelectedKeys(Set<SelectionKey> keys) throws Exception
    {
        Iterator<SelectionKey> itr = keys.iterator();
        while (itr.hasNext())
        {
            SelectionKey key = itr.next();
            if (key.isReadable())
                read(key);
            if (key.isWritable() && Server.syncWriteServerMap.get(skr) != 0) 
                write(key);
            if (key.isConnectable())
                connect(key);
            if (key.isAcceptable()) ;
            itr.remove();
        }
    }
    /**
     * Конфигурируем наше соединение.
     * @param channel - наш, созданный канал
     * @throws IOException
     */
    private void configureChannel(SocketChannel channel) throws IOException
    {
        channel.configureBlocking(false);
        channel.socket().setSendBufferSize(BUFFER_SIZE); // 65536 kb
        channel.socket().setReceiveBufferSize(BUFFER_SIZE); // 65536 kb
        channel.socket().setKeepAlive(true);
        channel.socket().setReuseAddress(true);
        channel.socket().setSoLinger(false, 0);
        channel.socket().setSoTimeout(0);
        channel.socket().setTcpNoDelay(true);
    }
    /**
     * Закрываем канал. Убиваем нить
     * @param key - ключ, связанный с каналом
     * @throws IOException
     */
    private void closeChannel(SelectionKey sk) throws IOException
    {
        LOG.info("Close server connect");
        SocketChannel socketChannel = (SocketChannel)sk.channel();
        if (socketChannel.isConnected())
        {
            socketChannel.close(); // закрываем канал
        }
        sk.cancel(); // удаляем из списка селектора
        terminate();// убиваем поток
    }
    private SocketAddress getAddress()
    {
        return address;
    }

    private void setAddress(SocketAddress address)
    {
        this.address = address;
    }
    private void terminate()
    {
        if (thread != null)
        {
            Server.syncReadClientPacketMap.remove(skr);
            Server.syncReadServerPacketMap.remove(skr);
            Server.syncWriteServerMap.remove(skr);
            Server.syncWriteClientMap.remove(skr);
            Thread.currentThread().interrupt();
        }
    }
}

Это сообщение отредактировал(а) acin - 13.12.2010, 23:05

1 Пользователей читают эту тему (1 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| Java: Работа с сетью \| Следующая тема »