Java Network

Материал из SEWiki
Перейти к: навигация, поиск

Что почитать

Thinking In Java Enterprise (русский перевод) → Сетевое программирование с Сокетами и Каналами Использование select для высокоскоростного сетевого взаимодействия

1. WhoAmI

import java.net.InetAddress;
import java.net.UnknownHostException;

public class WhoAmI {
	public static void main(String[] args) {
        if (args.length != 1) {
            System.err.println("Usage: WhoAmI MachineName");
            System.exit(1);
        }
		try {
	        InetAddress a = InetAddress.getByName(args[0]);
	        InetAddress b = InetAddress.getByName(null);
	        InetAddress c = InetAddress.getByName("localhost");
	        InetAddress d = InetAddress.getByName("127.0.0.1");	        
		    System.out.println(a);			
		    System.out.println(b + " " + c + " " + d);			
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
	}

}

2. EchoClient

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;


public class EchoClient {
	   public static void main(String[] args) throws IOException {
		      InetAddress addr = InetAddress.getByName(null);
		      System.out.println("addr = " + addr);
		      Socket socket = new Socket(addr, EchoServer.PORT);
		      // Помещаем все в блок try-finally, чтобы
		      // быть уверенным, что сокет закроется:
		      try {
		         System.out.println("socket = " + socket);
		         BufferedReader in = new BufferedReader(new InputStreamReader(socket
		               .getInputStream()));
		         // Вывод автоматически Output быталкивается PrintWriter'ом.
		         PrintWriter out = new PrintWriter(new BufferedWriter(
		               new OutputStreamWriter(socket.getOutputStream())), true);
		         for (int i = 0; i < 10; i++) {
		            out.println("howdy " + i);
		            String str = in.readLine();
		            System.out.println(str);
		         }
		         out.println("END");
		      }
		      finally {
		         System.out.println("closing...");
		         socket.close();
		      }
		   }
}

3. EchoServer

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;


public class EchoServer {
	   // Выбираем порт вне пределов 1-1024:
	   public static final int PORT = 8080;
	  
	   public static void main(String[] args) throws IOException {
	      ServerSocket s = new ServerSocket(PORT);
	      System.out.println("Started: " + s);
	      try {
	         // Блокирует до тех пор, пока не возникнет соединение:
	         Socket socket = s.accept();
	         try {
	            System.out.println("Connection accepted: " + socket);
	            BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
	            
	            // Вывод автоматически выталкивается из буфера PrintWriter'ом
	            PrintWriter out = new PrintWriter(new BufferedWriter(
	                  new OutputStreamWriter(socket.getOutputStream())), true);
	            while (true) {
	               String str = in.readLine();
	               if (str.equals("END"))
	                  break;
	               System.out.println("Echoing: " + str);
	               out.println(str);
	            }
	            // Всегда закрываем два сокета...
	         }
	         finally {
	            System.out.println("closing...");
	            socket.close();
	         }
	      }
	      finally {
	         s.close();
	      }
	   }
}

4. MultiEchoServer

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

class OneEchoServer extends Thread {
   private Socket socket;
   private BufferedReader in;
   private PrintWriter out;
  
   public OneEchoServer(Socket s) throws IOException {
      socket = s;
      in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
      // Включаем автоматическое выталкивание:
      out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())), true);
      // Если любой из вышеприведенных вызовов приведет к
      // возникновению исключения, то вызывающий отвечает за
      // закрытие сокета. В противном случае, нить
      // закроет его.
      start(); // вызываем run()
   }
  
   public void run() {
      try {
         while (true) {
            String str = in.readLine();
            if (str.equals("END"))
               break;
            System.out.println("Echoing: " + str);
            out.println(str);
         }
         System.out.println("closing...");
      }
      catch (IOException e) {
         System.err.println("IO Exception");
      }
      finally {
         try {
            socket.close();
         }
         catch (IOException e) {
            System.err.println("Socket not closed");
         }
      }
   }
}

public class MultiEchoServer {
   static final int PORT = 8080;
  
   public static void main(String[] args) throws IOException {
      ServerSocket s = new ServerSocket(PORT);
      System.out.println("Server Started");
      try {
         while (true) {
            // Блокируется до возникновения нового соединения:
            Socket socket = s.accept();
            try {
               new OneEchoServer(socket);
            }
            catch (IOException e) {
               // Если завершится неудачей, закрывается сокет,
               // в противном случае, нить закроет его:
               socket.close();
            }
         }
      }
      finally {
         s.close();
      }
   }
} // /:~


5. MultiEchoClient

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;

class EchoClientThread extends Thread {
   private Socket socket;
   private BufferedReader in;
   private PrintWriter out;
   private static int counter = 0;
   private int id = counter++;
   private static int threadcount = 0;
  
   public static int threadCount() {
      return threadcount;
   }
  
   public EchoClientThread(InetAddress addr) {
      System.out.println("Making client " + id);
      threadcount++;
      try {
         socket = new Socket(addr, MultiEchoServer.PORT);
      }
      catch (IOException e) {
         System.err.println("Socket failed");
         // Если создание сокета провалилось,
         // ничего ненужно чистить.
      }
      try {
         in = new BufferedReader(new InputStreamReader(socket
               .getInputStream()));
         // Включаем автоматическое выталкивание:
         out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())), true);
         start();
      }
      catch (IOException e) {
         // Сокет должен быть закрыт при любой
         // ошибке, кроме ошибки конструктора сокета:
         try {
            socket.close();
         }
         catch (IOException e2) {
            System.err.println("Socket not closed");
         }
      }
      // В противном случае сокет будет закрыт
      // в методе run() нити.
   }
  
   public void run() {
      try {
         for (int i = 0; i < 25; i++) {
            out.println("Client " + id + ": " + i);
            try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
            String str = in.readLine();
            System.out.println(str);
         }
         out.println("END");
      }
      catch (IOException e) {
         System.err.println("IO Exception");
      }
      finally {
         // Всегда закрывает:
         try {
            socket.close();
         }
         catch (IOException e) {
            System.err.println("Socket not closed");
         }
         threadcount--; // Завершаем эту нить
      }
   }
}

public class MultiEchoClient {
   static final int MAX_THREADS = 40;
  
   public static void main(String[] args) throws IOException,
         InterruptedException {
      InetAddress addr = InetAddress.getByName(null);
      while (true) {
         if (EchoClientThread.threadCount() < MAX_THREADS) {
            new EchoClientThread(addr);
         }
         Thread.sleep(100);
      }
   }
} // /:~


6. UDPEchoClient

import java.net.*; 

import java.io.*;

public class UDPEchoCLient{
	private InetAddress addr;
	private int port;
	private DatagramSocket ds;
	
	UDPEchoCLient(String host, int port) throws UnknownHostException, SocketException{ 
		this.addr = InetAddress.getByName(host);
		this.port = port; 
		this.ds = new DatagramSocket(); 
	}

	public void sendMessage(String mes){ 
		try{
			byte[] data = mes.getBytes();
			DatagramPacket pack = new DatagramPacket(data, data.length, addr, port); 
			ds.send(pack); 
		} catch(IOException e){
			System.err.println(e); 

		} 
	}
	
	public void close() {
		ds.close();
	}

	public static void main(String[] args) {
		try {
			UDPEchoCLient sndr = new UDPEchoCLient("localhost", 1050);
			for (String mes : args)
				sndr.sendMessage(mes);
			sndr.close();
		} catch (UnknownHostException | SocketException e) {
			e.printStackTrace();
		} 
	} 
}

7. UDPEchoServer

import java.net.DatagramPacket;
import java.net.DatagramSocket;


public class UDPEchoServer {
	public static void main(String[] args) {
		try {
			DatagramSocket ds = new DatagramSocket(1050); 

			while (true){
				DatagramPacket pack = new DatagramPacket(new byte[1024], 1024);
				ds.receive(pack);
				System.out.println(new String(pack.getData())); 
			} 
		} catch (Exception e) {
			System.out.println(e); 
		}  
	}
}

8. NonBlockingIO

import java.net.*;
import java.nio.channels.*;
import java.util.*;
import java.io.*;

/**
* Цель: Показать как использовать селектор. Нет чтения/записи, просто
* показывается готовность к совершению операции.
*
* Алгоритм: -> Создаем селектор. -> Создаем канал -> Связываем сокет,
* ассоциированный с каналом, с <клиентским портом> -> Конфигурируем канал, как
* не блокирующий -> Регестрируем канал в селекторе. -> Вызываем метод select( ),
* чтобы он блокировал выполнение до тех пор, пока канал не будет готов. (как
* это предполагается методом select(long timeout) -> Получаем множество ключей,
* относящихся к готовому каналу для работы, основной интерес состоит в том,
* когда они зарегестрированя с помощью селектора. -> Перебираем ключи. -> Для
* каждого ключа проверяем, что соответствующий канал готов к работе, в которой
* он заинтересован. -> Если он готов, печатаем сообщение о готовности.
*
* Примечание: -> Необходим запущенный MultiJabberServer на локальной машине. Вы
* запускаете его и соединяетесь с локальным MultiJabberServer -> Он может стать
* причиной исключения в MultiJabberServer, но это исключение ожидаемо.
*/
public class NonBlockingIO {
   public static void main(String[] args) throws IOException {
      if (args.length < 2) {
         System.out.println("Usage: java <client port> <local server port>");
         System.exit(1);
      }
      int cPort = Integer.parseInt(args[0]);
      int sPort = Integer.parseInt(args[1]);
      SocketChannel ch = SocketChannel.open();
      Selector sel = Selector.open();
      try {
         ch.socket().bind(new InetSocketAddress(cPort));
         ch.configureBlocking(false);
         // Канал заинтересован в выполнении чтения/записи/соединении
         ch.register(sel, SelectionKey.OP_READ | SelectionKey.OP_WRITE
               | SelectionKey.OP_CONNECT);
         // Разблокируем, когда готовы к чтению/записи/соединению
         sel.select();
         // Ключи, относящиеся к готовому каналу, канал заинтересован
         // в работе, которая может быть выполненаin can be
         // без блокирования.
         Iterator<?> it = sel.selectedKeys().iterator();
         while (it.hasNext()) {
            SelectionKey key = (SelectionKey) it.next();
            it.remove();
            // Если связанный с ключом канал готов к соединению?
            // if((key.readyOps() & SelectionKey.OP_CONNECT) != 0) {
            if (key.isConnectable()) {
               InetAddress ad = InetAddress.getLocalHost();
               System.out.println("Connect will not block");
               // Вы должны проверить возвращаемое значение,
               // чтобы убедиться, что он соединен. Этот не блокированный
               // вызов может вернуться без соединения, когда
               // нет сервера, к которому вы пробуете подключиться
               // Поэтому вы вызываете finishConnect(), который завершает
               // операцию соединения.
               if (!ch.connect(new InetSocketAddress(ad, sPort)))
                  ch.finishConnect();
            }
            // Если канал, связанный с ключом, готов к чтению?
            // if((key.readyOps() & SelectionKey.OP_READ) != 0)
            if (key.isReadable())
               System.out.println("Read will not block");
            // Готов ли канал, связанный с ключом, к записи?
            // if((key.readyOps() & SelectionKey.OP_WRITE) != 0)
            if (key.isWritable())
               System.out.println("Write will not block");
         }
      }
      finally {
         ch.close();
         sel.close();
      }
   }
} // /:~

9. EchoServerNIO

import java.io.*;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.charset.*;
import java.util.*;

/**
* Сервер принимает соединения не блокирующим способом. Когда соединение
* установлено, создается сокет, который регистрируется с селектором для
* чтения/записи. Чтение/запись выполняется над этим сокетом, когда селектор
* разблокируется. Эта программа работает точно так же, как и MultiJabberServer.
*/
public class EchoServerNIO {
   public static final int PORT = 8080;
  
   public static void main(String[] args) throws IOException {
      // Канал будет читать данные в ByteBuffer, посылаемые
      // методом PrintWriter.println(). Декодирование этого потока
      // байт требует кодовой страницы для кодировки по умолчанию.
      Charset cs = Charset.forName(System.getProperty("file.encoding"));
      ByteBuffer buffer = ByteBuffer.allocate(16);
      SocketChannel ch = null;
      ServerSocketChannel ssc = ServerSocketChannel.open();
      Selector sel = Selector.open();
      try {
         ssc.configureBlocking(false);
         // Локальныйы адрес, на котором он будет слушать соединения
         // Примечание: Socket.getChannel() возвращает null, если с ним не
         // ассоциирован канал, как показано ниже.
         // т.е выражение (ssc.socket().getChannel() != null) справедливо
         ssc.socket().bind(new InetSocketAddress(PORT));
         // Канал заинтересован в событиях OP_ACCEPT
         ssc.register(sel, SelectionKey.OP_ACCEPT);
         System.out.println("Server on port: " + PORT);
         while (true) {
            sel.select();
            Iterator<?> it = sel.selectedKeys().iterator();
            while (it.hasNext()) {
               SelectionKey skey = (SelectionKey) it.next();
               it.remove();
               if (skey.isAcceptable()) {
                  ch = ssc.accept();
                  System.out.println("Accepted connection from:"
                        + ch.socket());
                  ch.configureBlocking(false);
                  ch.register(sel, SelectionKey.OP_READ);
               }
               else {
                  // Обратите внимание, что не выполняется проверка, если
                  // в канал можно писать или читать - для упрощения.
                  ch = (SocketChannel) skey.channel();
                  ch.read(buffer);
                  CharBuffer cb = cs.decode((ByteBuffer) buffer.flip());
                  String response = cb.toString();
                  System.out.print("Echoing : " + response);
                  ch.write((ByteBuffer) buffer.rewind());
                  if (response.indexOf("END") != -1)
                     ch.close();
                  buffer.clear();
               }
            }
         }
      }
      finally {
         if (ch != null)
            ch.close();
         ssc.close();
         sel.close();
      }
   }
} // /:~

10. Worker

import java.io.*;
import java.util.logging.*;

public class Worker extends Thread {
   public static final Logger logger = Logger.getLogger("Worker");
   private String workerId;
   private Runnable task;
   // Необходима ссылка на пул нитей в котором существует нить, чтобы
   // нить могла добавить себя в пул нитей по завершению работы.
   private ThreadPool threadpool;
   static {
      try {
         logger.setUseParentHandlers(false);
         FileHandler ferr = new FileHandler("WorkerErr.log");
         ferr.setFormatter(new SimpleFormatter());
         logger.addHandler(ferr);
      }
      catch (IOException e) {
         System.out.println("Logger not initialized..");
      }
   }
  
   public Worker(String id, ThreadPool pool) {
      workerId = id;
      threadpool = pool;
      start();
   }
  
   // ThreadPool, когда ставит в расписание задачу, использует этот метод
   // для делегирования задачи Worker-нити. Кроме того для установки
   // задачи (типа Runnable) он также переключает ожидающий метод
   // run() на начало выполнения задачи.
   public void setTask(Runnable t) {
      task = t;
      synchronized (this) {
         notify();
      }
   }
  
   public void run() {
      try {
         while (!threadpool.isStopped()) {
            synchronized (this) {
               if (task != null) {
                  try {
                     task.run(); // Запускаем задачу
                  }
                  catch (Exception e) {
                     logger.log(Level.SEVERE,
                           "Exception in source Runnable task", e);
                  }
                  // Возвращает себя в пул нитей
                  threadpool.putWorker(this);
               }
               wait();
            }
         }
         System.out.println(this + " Stopped");
      }
      catch (InterruptedException e) {
         throw new RuntimeException(e);
      }
   }
  
   public String toString() {
      return "Worker : " + workerId;
   }
} // /:~

11. ThreadPool

import java.util.*;

public class ThreadPool extends Thread {
   private static final int DEFAULT_NUM_WORKERS = 5;
   private LinkedList<Worker> workerPool = new LinkedList<Worker>();
   private LinkedList<Runnable> taskList = new LinkedList<Runnable>();
   private boolean stopped = false;
  
   public ThreadPool() {
      this(DEFAULT_NUM_WORKERS);
   }
  
   public ThreadPool(int numOfWorkers) {
      for (int i = 0; i < numOfWorkers; i++)
         workerPool.add(new Worker("" + i, this));
      start();
   }
  
   public void run() {
      try {
         while (!stopped) {
            if (taskList.isEmpty()) {
               synchronized (taskList) {
                  // Если очередь пустая, подождать, пока будет добавлена
                  // задача
                  taskList.wait();
               }
            }
            else if (workerPool.isEmpty()) {
               synchronized (workerPool) {
                  // Если нет рабочих нитей, подождать, пока
                  // пока не появится
                  workerPool.wait();
               }
            }
            // Запускаем следующую задачу из расписания задач
            getWorker().setTask((Runnable) taskList.removeLast());
         }
      }
      catch (InterruptedException e) {
         throw new RuntimeException(e);
      }
   }
  
   public void addTask(Runnable task) {
      taskList.addFirst(task);
      synchronized (taskList) {
         taskList.notify(); // Если добавлена новая задача, уведомляем
      }
   }
  
   public void putWorker(Worker worker) {
      workerPool.addFirst(worker);
      // Здесь может быть случай, когда вы будете иметь пул из 5 нитей,
      // а будет требоваться больше. Это происходит тогда, когда требуется
      // рабочая нить,
      // но ее нет (свободной), тогда просто блокируем пул нитей.
      // Это событие, при котором появляется свободная рабочая нить в пуле
      // нитей
      // Поэтому эта нить посылает уведомление и разблокирует
      // нить ThreadPool, ожидающую пул нитей
      synchronized (workerPool) {
         workerPool.notify();
      }
   }
  
   private Worker getWorker() {
      return (Worker) workerPool.removeLast();
   }
  
   public boolean isStopped() {
      return stopped;
   }
  
   public void stopThreads() {
      stopped = true;
      Iterator<Worker> it = workerPool.iterator();
      while (it.hasNext()) {
         Worker w = it.next();
         synchronized (w) {
            w.notify();
         }
      }
   } // Junit test
  
   public void testThreadPool() {
      ThreadPool tp = new ThreadPool();
      for (int i = 0; i < 10; i++) {
         tp.addTask(new Runnable() {
            public void run() {
               System.out.println("A");
            }
         });
      }
      tp.stopThreads();
   }
} // /:~

12. MultiEchoServerPool

import java.io.*;

import java.net.*;

import java.nio.*;

import java.nio.channels.*;

import java.nio.charset.*;

import java.util.*;

class ServeOneEcho implements Runnable {
   private SocketChannel channel;
   private Selector sel;
  
   public ServeOneEcho(SocketChannel ch) throws IOException {
      channel = ch;
      sel = Selector.open();
   }
  
   public void run() {
      ByteBuffer buffer = ByteBuffer.allocate(16);
      boolean read = false, done = false;
      String response = null;
      try {
         channel.register(sel, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
         while (!done) {
            sel.select();
            Iterator<?> it = sel.selectedKeys().iterator();
            while (it.hasNext()) {
               SelectionKey key = (SelectionKey) it.next();
               it.remove();
               if (key.isReadable() && !read) {
                  if (channel.read(buffer) > 0)
                     read = true;
                  CharBuffer cb = MultiEchoServerPool.CS
                        .decode((ByteBuffer) buffer.flip());
                  response = cb.toString();
               }
               if (key.isWritable() && read) {
                  System.out.print("Echoing : " + response);
                  channel.write((ByteBuffer) buffer.rewind());
                  if (response.indexOf("END") != -1)
                     done = true;
                  buffer.clear();
                  read = false;
               }
            }
         }
      }
      catch (IOException e) {
         // будет поймано Worker.java и залогировано.
         // Необходимо выбросить исключение времени выполнения, так как мы не
         // можем
         // оставить IOException
         throw new RuntimeException(e);
      }
      finally {
         try {
            channel.close();
         }
         catch (IOException e) {
            System.out.println("Channel not closed.");
            // Выбрасываем это, чтобы рабочая нить могла залогировать.
            throw new RuntimeException(e);
         }
      }
   }
}

public class MultiEchoServerPool {
   public static final int PORT = 8080;
   private static String encoding = System.getProperty("file.encoding");
   public static final Charset CS = Charset.forName(encoding);
   // Создаем пул нитей с 20 рабочими нитями.
   private static ThreadPool pool = new ThreadPool(20);
  
   public static void main(String[] args) throws IOException {
      ServerSocketChannel ssc = ServerSocketChannel.open();
      Selector sel = Selector.open();
      try {
         ssc.configureBlocking(false);
         ssc.socket().bind(new InetSocketAddress(PORT));
         ssc.register(sel, SelectionKey.OP_ACCEPT);
         System.out.println("Server on port: " + PORT);
         while (true) {
            sel.select();
            Iterator<?> it = sel.selectedKeys().iterator();
            while (it.hasNext()) {
               SelectionKey skey = (SelectionKey) it.next();
               it.remove();
               if (skey.isAcceptable()) {
                  SocketChannel channel = ssc.accept();
                  System.out.println("Accepted connection from:"
                        + channel.socket());
                  channel.configureBlocking(false);
                  // Отделяем события и ассоциированное действие
                  pool.addTask(new ServeOneEcho(channel));
               }
            }
         }
      }
      finally {
         ssc.close();
         sel.close();
      }
   }
} // /:~
Источник — «http://mit.spbau.ru/sewiki/index.php?title=Java_Network&oldid=1331»