Java Socket分发服务负载均衡

  1 1、 设备请求分发服务器，分发服务器返回有效的socket服务器ip与port，然后断开连接。
  2 a) 设备与服务器建立连接。
  3 b) 服务器接收到连接请求后，立即将分配好的socket服务器ip与port信息响应给设备。
  4 c) 服务器主动断开socket连接。
  5 2、 设备得到ip与port以后，设备去连接socket服务器，然后与其进行协议通讯。
  6 a) 设备连接到socket服务器。
  7 b) socket服务器响应连接成功响应信息。
  8 c) 设备与socket服务器保持长链接通讯。
  9 
 10 *. 若设备未收到连接成功响应则再次尝试连接，若三次请求依旧没有成功建立连接，那么设备需要去请求分发服务器然后再重新上述操作。
 11 *. 当设备在异常情况下链接不上socket服务器时，依旧尝试三次若不能成功，则直接请求分发服务器，然后再重复上述操作。
 12 
 13 分发服务器代码：
 14 分发服务与客户端是短链接：
 15 package com.easygo.server;
 16 
 17 import java.io.IOException;
 18 import java.net.ServerSocket;
 19 import java.net.Socket;
 20 import java.util.concurrent.ExecutorService;
 21 import java.util.concurrent.Executors;
 22 
 23 public class CenterServer implements Runnable{
 24     ServerSocket serverSocket=null;
 25     static volatile String server_ip="127.0.0.1";
 26     static volatile int serverport1=8085;
 27     static volatile int serverport2=8086;
 28     public static void main(String[] args) {
 29         new Thread(new CenterServer()).start();
 30         
 31     }
 32 
 33     @Override
 34     public void run() {
 35          ExecutorService cachedThreadPool = null;//线程池
 36             cachedThreadPool = Executors.newCachedThreadPool();//线程池
 37             try {
 38                 serverSocket=new ServerSocket(9999);
 39             } catch (IOException e1) {
 40                 // TODO Auto-generated catch block
 41                 e1.printStackTrace();
 42             }
 43         while(true){
 44             try {
 45                 
 46                 cachedThreadPool.execute(new ServerRun(serverSocket.accept()));
 47                 System.out.println("有一个客户端连进来了..");
 48             } catch (IOException e) {
 49                 // TODO Auto-generated catch block
 50                 e.printStackTrace();
 51             }
 52             
 53         }
 54     }
 55 
 56     
 57     
 58 }
 59 class ServerRun implements Runnable{
 60     Socket socket=null;
 61     ServerRun(Socket socket){
 62         this.socket=socket;
 63         
 64     }
 65     @Override
 66     public void run() {
 67         
 68         receive();
 69         
 70         
 71     }
 72     public void send(String message){
 73         try {
 74             socket.getOutputStream().write(message.getBytes());
 75             socket.getOutputStream().flush();
 76         } catch (IOException e) {
 77             // TODO Auto-generated catch block
 78             e.printStackTrace();
 79         }
 80         
 81         
 82     }
 83     public void receive(){
 84         
 85         byte[]tmp=null;
 86         String message=null;
 87         
 88         boolean runs=true;
 89         while(runs){
 90             if(null!=socket){
 91                 tmp=new byte[30];
 92                 
 93                 @SuppressWarnings("unused")
 94                 int count =0;
 95                 try {
 96                     while((count=socket.getInputStream().read(tmp))!=-1){
 97                         message=new String(tmp);
 98                         System.out.println("Clent:"+message.trim());
 99                         if(message.trim().split(",")[0].equals("login")&&message.trim().split(",")[1].equals("1")){
100                             
101                             send("login,"+CenterServer.server_ip+","+CenterServer.serverport1);
102                         }else if(message.trim().split(",")[0].equals("login")&&message.trim().split(",")[1].equals("2")){
103                             
104                             send("login,"+CenterServer.server_ip+","+CenterServer.serverport2);
105                         }else if(message.equals("0")){
106                             if(socket!=null){
107                                 socket.close();
108                                 
109                             }
110                             
111                             
112                         }
113                         if(socket.isClosed()){
114                             runs=false;
115                         }
116                         tmp=null;
117                         message=null;
118                         tmp=new byte[30];
119                     }
120                     
121                     
122                 } catch (IOException e) {
123                     
124                 }
125                 
126             }
127             
128             
129         }
130     }
131     
132 }
133 
134 
135 
136 客户端代码案例：
137 
138 package com.easygo.server;
139 
140 
141 import java.io.IOException;
142 import java.net.Socket;
143 
144 
145 public class Client implements Runnable{
146     Socket socket=null;
147     final int CENTER_SERVER_PORT=9999;//分发服务器端口
148     final String CENTER_SERVER_IP="127.0.0.1";
149      int CENTER_SERVER_PORTS;
150      String CENTER_SERVER_IPS;
151     boolean conn=false;
152     volatile boolean receive;
153 public static void main(String[] args) {
154     new Thread(new Client()).start();
155 }
156 
157 @Override
158 public void run() {
159     try {
160         connection();
161     } catch (Exception e) {
162         // TODO Auto-generated catch block
163         e.printStackTrace();
164     }
165     
166     
167 }
168 
169 public void connection() throws Exception{
170     
171     while(true){
172         if(null==socket&&conn==false){//登入分发服务器
173             socket=new Socket(CENTER_SERVER_IP, CENTER_SERVER_PORT);
174             receive=true;
175             
176             send("login,1");//login,登入分发服务器，1指定登入服务
177                     
178             
179             new Thread(){
180                 public void run(){
181                     try {
182                         receiveX();
183                     } catch (IOException e) {
184                         // TODO Auto-generated catch block
185                         e.printStackTrace();
186                     }
187                     
188                 }
189                 
190             }.start();
191         }else if(null==socket&&conn==true){//登入指定服务
192             socket=new Socket(CENTER_SERVER_IPS, CENTER_SERVER_PORTS);
193             receive=true;
194             new Thread(){
195                 @Override
196                 public void run(){
197                     while(true){
198                         send("message,我是客户端");
199                         
200                         try {
201                             Thread.sleep(1000);
202                         } catch (InterruptedException e) {
203                             // TODO Auto-generated catch block
204                             e.printStackTrace();
205                         }
206                     }
207                     
208                     
209                 }
210                 
211             }.start();
212             new Thread(){
213                 public void run(){
214                     try {
215                         receiveX();
216                     } catch (IOException e) {
217                         // TODO Auto-generated catch block
218                         e.printStackTrace();
219                     }
220                     
221                 }
222                 
223             }.start();
224         }
225         
226         Thread.sleep(1000);
227         
228     }
229 }
230 
231 public void send(String message){
232     System.out.println("send:"+message);
233     try {
234         socket.getOutputStream().write(message.getBytes());
235         socket.getOutputStream().flush();
236     } catch (IOException e) {
237     
238     }
239     
240 }
241 
242 public void receiveX() throws IOException{
243     byte[]tmp=null;
244     String messaage=null;
245     while(receive){
246         if(null!=socket){
247             tmp=new byte[50];
248             @SuppressWarnings("unused")
249             int count=0;
250             while((count=socket.getInputStream().read(tmp))!=-1){
251                 messaage=new String(tmp);
252                 System.out.println("result:"+messaage);
253                 if(messaage.trim().contains("login")){
254                     String[]arr=messaage.trim().split(",");
255                     System.out.println(":"+arr[1]+":"+arr[2]);
256                     CENTER_SERVER_IPS=arr[1];
257                     CENTER_SERVER_PORTS=Integer.parseInt(arr[2]);
258                     
259                     
260                     conn=true;
261                     //send("0");
262                     closeXX();//获取到对应服务ip和端口后断开连接
263                     receive=false;//
264                     
265                     
266                 }else if(messaage.trim().contains("message")){
267                     send(messaage);
268                     
269                 }
270                 tmp=null;
271                 tmp=new byte[50];
272             }
273             
274             
275             
276             
277         }
278         tmp=null;
279         messaage=null;
280     }
281     
282 }
283 public void closeXX(){
284     if(this.socket!=null){
285         try {
286             socket.close();
287             socket=null;
288         } catch (IOException e) {
289             // TODO Auto-generated catch block
290             e.printStackTrace();
291         }
292         
293     }
294 }
295 }
296 
297 
298 指定登入处理客户端服务代码
299 package com.easygo.server;
300 
301 import java.io.IOException;
302 import java.net.ServerSocket;
303 import java.net.Socket;
304 import java.util.concurrent.ExecutorService;
305 import java.util.concurrent.Executors;
306 
307 public class Server implements Runnable{
308     volatile int port=8086;
309     ServerSocket serverSocket=null;
310     public static void main(String[] args) {
311         new Thread(new Server()).start();
312     }
313 
314     @Override
315     public void run() {
316         int count=0;
317         try {
318             serverSocket=new ServerSocket(port);
319             ExecutorService cachedThreadPool = null;//线程池
320             cachedThreadPool = Executors.newCachedThreadPool();//线程池
321             
322             while(true){
323                 cachedThreadPool.execute(new ServerRunS(serverSocket.accept()));
324                 System.out.println("a client conn:"+count++);
325                 
326             }
327         } catch (IOException e) {
328             // TODO Auto-generated catch block
329             e.printStackTrace();
330         }
331         
332     }
333 
334 }
335 class ServerRunS implements Runnable{
336     Socket socket=null;
337     ServerRunS(Socket socket){
338         
339         this.socket=socket;
340     }
341     @Override
342     public void run() {
343         byte[]tmp=null;
344         
345         while(true){
346             
347             if(socket!=null){
348                 tmp=new byte[30];
349                 @SuppressWarnings("unused")
350                 int count=0;
351                 try {
352                     while((count=socket.getInputStream().read(tmp))!=-1){
353                     System.out.println("客户端发来的消息："+new String(tmp));
354                         
355                     }
356                 } catch (IOException e) {
357                     // TODO Auto-generated catch block
358                     e.printStackTrace();
359                 }
360                 
361             }
362             tmp=null;
363         }
364     }
365     
366     
367     
368 }
369 
370 流程：
371 1启动server
372 2启动CenterServer服务
373 3启动客户端一，参数（login,1）
374 4启动客户端二，参数（login,2）
375 
376 运行结果：
377 客户端1
378 send:login,1
379 result:login,127.0.0.1,8085
380 send:message,我是客户端
381 send:message,我是客户端
382 send:message,我是客户端
383 send:message,我是客户端
384 send:message,我是客户端
385 send:message,我是客户端
386 
387 客户端2
388 send:login,2
389 result:login,127.0.0.1,8086
390 send:message,我是客户端
391 send:message,我是客户端
392 send:message,我是客户端
393 send:message,我是客户端
394 send:message,我是客户端
395 send:message,我是客户端
396 
397 分发服务器：
398 有一个客户端连进来了
399 UUlogin,2
400 有一个客户端连进来了
401 UUlogin,1
402 
403 服务器1（8085端口）
404 8085
405 客户端发来的消息：message,我是客户端
406 客户端发来的消息：message,我是客户端
407 .
408 .
409 .
410 服务器1（8086端口）
411 8086
412 客户端发来的消息：message,我是客户端
413 客户端发来的消息：message,我是客户端
414 .
415 .
416 .