Цитата(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();
}
}
}
|
|