1.为了安全的进行通信，基于ACK确认机制----重传机制
使用selector机制的多路复用的IO通信机制，此外还有epoll控制
业务挂接点或者接入点：
常见的接入点：
RDBMS进行CRUD
MQ消息服务器接入
SOA接入点
如果统一，最好使用adapter适配器接入，方便扩展
package com.text.thread; import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level; public class SocketUtil { ExecutorService excu = Executors.newFixedThreadPool(100);
private volatile boolean status = true; void startServer(int port, String message) {
try {
ServerSocket serverSocket = new ServerSocket(port);
if (status) {
serverSocket.close();
} while (status) {
Socket socket = serverSocket.accept();
Runnable task = new ServerTask(socket, message);
excu.submit(task);
}
} catch (IOException e) {
e.printStackTrace();
} finally {
}
} void startNioServer(int port, String messages) {
Selector selector = null;
ServerSocketChannel serverSocketChannel = null; try {
selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false); serverSocketChannel.socket().setReuseAddress(true);
serverSocketChannel.socket().bind(new InetSocketAddress(port)); serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); while (selector.select() > 0) {
Iterator<SelectionKey> it = selector.selectedKeys().iterator(); while (it.hasNext()) {
SelectionKey readyKey = it.next();
it.remove(); ServerSocketChannel sChannel = (ServerSocketChannel) readyKey.channel();
SocketChannel schannel = sChannel.accept(); String messgae = "";
ByteBuffer dsts = ByteBuffer.allocate(256);
int flag = -1;
while((flag = schannel.read(dsts))!=-1){
String msg = new String(dsts.array(),"GBK");
messgae+="\n" + msg;
} System.out.println("server:" + messgae);//此处就是具体的业务挂节点或者接入点--比如执行DB操作或者执行MQ操作
ByteBuffer bf = ByteBuffer.wrap(messages.getBytes());
schannel.write(bf); schannel.close();
}
}
} catch (Exception e) { }
} void requestNio(String host,int port,String msg){ SocketChannel socketChannel;
try {
socketChannel = SocketChannel.open();
SocketAddress socketAddress = new InetSocketAddress(host, port);
socketChannel.connect(socketAddress); ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
socketChannel.write(buffer);
socketChannel.socket().shutdownOutput(); ByteBuffer dsts = ByteBuffer.allocate(100);
socketChannel.read(dsts); System.out.println(new String(dsts.array())); socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
} } private void executeTask(ServerSocketChannel serverSocketChannel,String msg) {
SocketChannel socketChannel = null;
try {
socketChannel = serverSocketChannel.accept(); socketChannel.socket().shutdownOutput(); } catch (IOException e) {
e.printStackTrace();
} finally {
try {
socketChannel.close();
} catch(Exception ex) {}
}
} void stopServer() {
status = false;
} String request(String ip, int port, String message) {
String result = "";
try {
Socket socket = new Socket(ip, port); OutputStream out = socket.getOutputStream();
OutputStreamWriter outputStreamWriter = new OutputStreamWriter(out);
BufferedWriter bufferedWriter = new BufferedWriter(
outputStreamWriter); bufferedWriter.write("new Msg");// 换行
bufferedWriter.newLine();// 换行
bufferedWriter.write(message);
bufferedWriter.newLine();// 换行
bufferedWriter.flush();
socket.shutdownOutput();
System.out.println("client:发送成功!!!"); InputStream in = socket.getInputStream();
InputStreamReader inputStreamReader = new InputStreamReader(in);
BufferedReader bufferedReader = new BufferedReader(
inputStreamReader); System.out.println("client:" + bufferedReader.readLine());
result = bufferedReader.readLine();
bufferedReader.close();
bufferedWriter.close(); } catch (UnknownHostException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return result;
} static class ServerTask implements Runnable { Socket socket;
String message; public ServerTask(Socket socket, String message) {
this.socket = socket;
this.message = message;
} @Override
public void run() {
InputStream in;
try {
in = socket.getInputStream();
InputStreamReader inputStreamReader = new InputStreamReader(in);
BufferedReader bufferedReader = new BufferedReader(
inputStreamReader); String temp = "";
String msg = "";
String str = bufferedReader.readLine();
if (str != null && !str.equals("new Msg")) {
System.out.println("此消息已经接收!!!");
temp = "!已经接受过";
} else { while (str != null) {
msg += str + "\n";
str = bufferedReader.readLine();
}
System.out.println("server:\n" + msg);
}
OutputStream out = socket.getOutputStream();
OutputStreamWriter outputStreamWriter = new OutputStreamWriter(
out);
BufferedWriter bufferedWriter = new BufferedWriter(
outputStreamWriter); bufferedWriter.write(message + temp);
bufferedWriter.newLine();// 换行
bufferedWriter.flush(); System.out.println("server:完成");
bufferedWriter.close();
bufferedReader.close();
} catch (IOException e) {
e.printStackTrace();
} } } }
SocketUtil socketUtil = new SocketUtil();
String message = "1001-102401-骨科一病区-李四-20140908|20140925-3786.98元!";
socketUtil.request("127.0.0.1", 9999, message );
SocketUtil socketUtil = new SocketUtil();
socketUtil.startServer( 9999, "ack"); 消息：（协议自己按照业务需求进行定义，只要双方可以交流通信即可！！！）
MSH|^~\&|LIS||HIS|DEP|201106271801||ACK^O13|MSG11O1O5OOO001|P|2.4|| MSG11O1O5OOO001||||utf-8
MSA|AE|MSG11O1O5OOO001||||103
MSH|^~\&|LIS||HIS|DEP|201106271801||ACK^O13|MSG11O1O5OOO001|P|2.4|| MSG11O1O5OOO001||||utf-8
MSA|AA|MSG11O1O5OOO001||||0
对于文件的传输，把文件压缩进行传输，为了安全，可以生成摘要信息，采取MD5或者RAS加密传输更加安全
对于大文件，分割多线程传输，采取Random读取文件的方式
服务器端接受到请求，必须发送收到的标识码，否则客户重新发送,加一个重发的标示符