java Socket长链接与消息推送源码与演示

演示视频如下(在服务端输入消息,客户端接收消息) : 

视频链接

---

服务端代码 MyServer.java 源码如下

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Scanner;


public class MyServer {
    
    private static final Map<String,Socket> connectMap = new HashMap<String,Socket>(); //链接信息
    
    private static ServerSocket server = null ;
    
    private static final int PORT = 8888 ;
    
    //SOCKET 服务线程类
    static class ReciveService extends Thread{
        
        Socket client = null ; 
        
        private InputStreamReader is = null;
        private BufferedReader in = null; 
        
        public ReciveService(Socket client){
            this.client = client ;
        }

        public void run() {

            long tid =Thread.currentThread().getId() ;//当前线程ID
            try{
                String IP = client.getInetAddress().getHostAddress() ;
                
                System.out.println("已经链接 : "+IP+"["+tid+"]");
                connectMap.put(IP+"#"+tid,client);
                is = new InputStreamReader(client.getInputStream(), "UTF-8");//规定编码
                in = new BufferedReader(is);//接收请求的流
                
                int len = 0;//监听到的字符串长度
                char[] buf = new char[1024];
                while ((len = in.read(buf)) != -1) {
                    String s = new String(buf, 0, len);
                    s = new String(s.getBytes(),"utf-8");
                    System.out.println("-> "+s);
                }
            }catch(Exception e){
                e.printStackTrace() ;
            }finally{
                //关闭Socket
                if(null != client){
                    try {
                        client.close() ;
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        
        }
        
        
    }
    
    //向客户端发送消息
    private static Thread tSend = new Thread(new Runnable(){
        Scanner scanner =new Scanner(System.in);
        public void run() {
            try {
                while (true) {
                    String data = scanner.next() ;
                    for(Iterator<String> it = connectMap.keySet().iterator();it.hasNext();){
                        String key = it.next() ;//IP+线程ID
                        Socket client = connectMap.get(key);
                        if(null != client && client.isConnected()){
                            OutputStream os= client.getOutputStream();            
                            PrintWriter pw=new PrintWriter(os);//转化为打印流  
                            pw.write(data);  
                            pw.flush();//刷新缓存
                        }
                    }
                    Thread.sleep(1000);
//                    System.out.println("...");
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
    
    
    
    
    public static void main(String[] args) throws IOException {
        System.out.println("==============SOCKET 服务端===============");
        startServer() ;
    }
    
    public static void startServer() throws IOException{
        server = new ServerSocket(PORT);
        tSend.start();                                            //启动发送消息线程
        while(true){
            try{
                Socket client = server.accept() ;                 //主线程获取客户端连接
                ReciveService service = new ReciveService(client) ;
                service.start();                                //启动接收消息线程
            }catch(Exception e){
                e.printStackTrace() ;
            }
        }
    }
}

客户端代码 MyClient.java 源码如下 :

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Scanner;

public class MyClient {

    private static final ThreadLocal<Socket> threadConnect = new ThreadLocal<Socket>(); 
    
    private static final String HOST = "localhost";

    private static final int PORT = 8888;
    
    private static Socket client;
    
    private static OutputStream outStr = null;
    
    private static InputStream inStr = null;
    
    //数据接收线程
    private static Thread tRecv = new Thread(new Runnable(){
        public void run() {
            try {
//              System.out.println("==============开始接收数据===============");
                while (true) {
                    byte[] b = new byte[1024];
                    int r = inStr.read(b);
                    if(r>-1){
                        String str = new String(b,0,r,"utf-8");
                        System.out.println( str );
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    });
    
    //消息发送线程
    private static Thread tSend = new Thread(new Runnable(){
        Scanner scanner =new Scanner(System.in);
        public void run() {
//            System.out.println("=====================开始发送数据==============");
            try {
                while (true) {
                    String data = scanner.next() ;
                    outStr.write(data.getBytes());
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
    });
    
    //心跳包发送线程
    private static Thread tKeep = new Thread(new Runnable(){
        public void run() {
            try {
//              System.out.println("=====================开始发送心跳包==============");
                while (true) {
                    try {
                        //http://blog.csdn.net/qq_23167527/article/details/54290726
                        Thread.sleep(30000);//心跳频率 30秒 
//                      System.out.println(".");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                  //发送一个心跳数据包
                    outStr.write(new byte[]{});//发送非空包也可以
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    });
    
    /**
     * @param args
     */
    public static void main(String[] args) {
        System.out.println("==============SOCKET 客户端===============");
        
        try {
            connect();
            tRecv.start() ;
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        
    }
    
    public static void connect() throws UnknownHostException, IOException  {
        client = threadConnect.get();
        if(client == null){
            client = new Socket(HOST, PORT);
            threadConnect.set(client);
            tKeep.start();//启动长连接心跳保活线程
            tSend.start() ;
//          System.out.println("========链接开始！========");
        }
        outStr = client.getOutputStream();
        inStr = client.getInputStream();
    }
    
}