netty-socketio 示例代碼

socket.io是一個不錯的websocket項目，github上有它的java實現：netty-socketio 及 示例項目 netty-socketio-demo，基本上看看demo示例項目就能很快上手了，但是demo中的示例代碼場景為js做客戶端，如果需要在java中連接websocket server，可以參考下面的示例:

一、服務端代碼

package com.corundumstudio.socketio.demo.server;

import com.corundumstudio.socketio.AckRequest;
import com.corundumstudio.socketio.Configuration;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.listener.ConnectListener;
import com.corundumstudio.socketio.listener.DataListener;
import io.socket.client.Socket;

/**
 * Created by yangjunming on 2017/1/13.
 */
public class DemoSocketServer {


    public static void main(String[] args) throws InterruptedException {

        Configuration config = new Configuration();
        config.setHostname("localhost");
        config.setPort(9092);

        final SocketIOServer server = new SocketIOServer(config);

        server.addConnectListener(new ConnectListener() {
            @Override
            public void onConnect(SocketIOClient client) {
                String token = client.getHandshakeData().getUrlParams().get("token").get(0);
                if (!token.equals("87df42a424c48313ef6063e6a5c63297")) {
                    client.disconnect();//校驗token示例
                }
                System.out.println("sessionId:" + client.getSessionId() + ",token:" + token);
            }
        });


        server.addEventListener(Socket.EVENT_MESSAGE, String.class, new DataListener<String>() {
            @Override
            public void onData(SocketIOClient client, String data, AckRequest ackSender) throws Exception {
                System.out.println("client data:" + data);
                server.getBroadcastOperations().sendEvent(Socket.EVENT_MESSAGE, "hi");
            }
        });

        server.start();
        Thread.sleep(Integer.MAX_VALUE);
        server.stop();
    }

}

服務端的主要工作，就是添加各種事件的監聽，然后在監聽處理中，做相應的處理即可。

注：添加事件監聽時，如果重復添加監聽，會導致事件被處理多次，所以最好在添加事件監聽前，先移除之前已經存在的監聽，類似下面這樣

        chat1namespace.removeAllListeners(Socket.EVENT_MESSAGE);
        chat1namespace.addEventListener(Socket.EVENT_MESSAGE, String.class,...

　　

 

二、客戶端代碼

java連接netty-socketio，還要借助另一個開源項目:socket.io-client-java

package com.corundumstudio.socketio.demo.client;

import io.socket.client.IO;
import io.socket.client.Socket;
import io.socket.emitter.Emitter;

import java.net.URISyntaxException;

/**
 * Created by yangjunming on 2017/1/13.
 */
public class DemoSocketClient {

    public static void main(String[] args) throws URISyntaxException, InterruptedException {
        IO.Options options = new IO.Options();
        options.transports = new String[]{"websocket"};
        options.reconnectionAttempts = 2;
        options.reconnectionDelay = 1000;//失敗重連的時間間隔
        options.timeout = 500;//連接超時時間(ms)

//        final Socket socket = IO.socket("http://localhost:9092/?token=123456", options);//錯誤的token值連接示例
        final Socket socket = IO.socket("http://localhost:9092/?token=87df42a424c48313ef6063e6a5c63297", options);

        socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                socket.send("hello");
            }
        });

        socket.on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                System.out.println("連接關閉");
            }
        });

        socket.on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                System.out.println("sessionId:" + socket.id());
                for (Object obj : args) {
                    System.out.println(obj);
                }
                System.out.println("收到服務器應答，將要斷開連接...");
                socket.disconnect();
            }
        });
        socket.connect();
    }
}

客戶端類似，也是加一些事件監聽，然后做相應處理即可。

 

上面的例子，演示了client向server連接時，如何做基本的連接認證(基於token），以及基本的消息收發。

運行效果：

服務端輸出

sessionId:f52e9fa3-6216-4742-87de-3228a74469f9,token:87df42a424c48313ef6063e6a5c63297
client data:hello

客戶端輸出

sessionId:f52e9fa3-6216-4742-87de-3228a74469f9
hi
收到服務器應答，將要斷開連接...
連接關閉

注：框架已經自帶了一些預設的事件，見下面的代碼片段

    /**
     * Called on a successful connection.
     */
    public static final String EVENT_OPEN = "open";

    /**
     * Called on a disconnection.
     */
    public static final String EVENT_CLOSE = "close";

    public static final String EVENT_PACKET = "packet";
    public static final String EVENT_ERROR = "error";

    /**
     * Called on a connection error.
     */
    public static final String EVENT_CONNECT_ERROR = "connect_error";

    /**
     * Called on a connection timeout.
     */
    public static final String EVENT_CONNECT_TIMEOUT = "connect_timeout";

    /**
     * Called on a successful reconnection.
     */
    public static final String EVENT_RECONNECT = "reconnect";

    /**
     * Called on a reconnection attempt error.
     */
    public static final String EVENT_RECONNECT_ERROR = "reconnect_error";

    public static final String EVENT_RECONNECT_FAILED = "reconnect_failed";

    public static final String EVENT_RECONNECT_ATTEMPT = "reconnect_attempt";

    public static final String EVENT_RECONNECTING = "reconnecting";

    public static final String EVENT_PING = "ping";

    public static final String EVENT_PONG = "pong";

如果不夠的話，可以自行擴展，無非就是一些字符串常量。　　

 

三、廣播消息隔離

前面的示例，沒有"域"的概念，所有連到socket server上的client，如果收發廣播的話，全都能收到，如果只希望將消息發到指定的某一"批"用戶，可以讓這些client歸到某個域(或組織機構)里，這樣在指定的域范圍內廣播，只有在這個域內的client才能接受廣播，詳見下面的示例:（其實變化很小）

server端：

package com.corundumstudio.socketio.demo.server;

import com.corundumstudio.socketio.*;
import com.corundumstudio.socketio.listener.ConnectListener;
import com.corundumstudio.socketio.listener.DataListener;
import io.socket.client.Socket;

/**
 * Created by yangjunming on 2017/1/13.
 */
public class DemoSocketServer {


    public static void main(String[] args) throws InterruptedException {
        SocketIOServer server = getServer();
        addRoom(server);
        startServer(server);
    }

    private static Configuration getConfig() {
        Configuration config = new Configuration();
        config.setHostname("localhost");
        config.setPort(9092);
        return config;
    }

    private static void handleConn(SocketIOServer server) {
        server.addConnectListener(new ConnectListener() {
            @Override
            public void onConnect(SocketIOClient client) {
                String token = client.getHandshakeData().getUrlParams().get("token").get(0);
                if (!token.equals("87df42a424c48313ef6063e6a5c63297")) {
                    client.disconnect();//校驗token示例
                }
                System.out.println("sessionId:" + client.getSessionId() + ",token:" + token);
            }
        });
    }

    private static void addRoom(SocketIOServer server) {
        final SocketIONamespace chat1namespace = server.addNamespace("/room1");
        chat1namespace.addEventListener(Socket.EVENT_MESSAGE, String.class, new DataListener<String>() {
            @Override
            public void onData(SocketIOClient client, String data, AckRequest ackRequest) {
                chat1namespace.getBroadcastOperations().sendEvent(Socket.EVENT_MESSAGE, "ack:" + data);
            }
        });
    }

    private static SocketIOServer getServer() throws InterruptedException {
        final SocketIOServer server = new SocketIOServer(getConfig());
        handleConn(server);

        server.addEventListener(Socket.EVENT_MESSAGE, String.class, new DataListener<String>() {
            @Override
            public void onData(SocketIOClient client, String data, AckRequest ackSender) throws Exception {
                System.out.println("client data:" + data);
                server.getBroadcastOperations().sendEvent(Socket.EVENT_MESSAGE, "hi");
            }
        });
        return server;
    }

    private static void startServer(SocketIOServer server) throws InterruptedException {
        server.start();
        Thread.sleep(Integer.MAX_VALUE);
        server.stop();
    }

}

客戶端：

package com.corundumstudio.socketio.demo.client;

import io.socket.client.IO;
import io.socket.client.Socket;
import io.socket.emitter.Emitter;

import java.net.URISyntaxException;

/**
 * Created by yangjunming on 2017/1/13.
 */
public class DemoSocketClient {

    public static void main(String[] args) throws URISyntaxException, InterruptedException {
        IO.Options options = new IO.Options();
        options.transports = new String[]{"websocket"};
        options.reconnectionAttempts = 2;
        options.reconnectionDelay = 1000;//失敗重連的時間間隔
        options.timeout = 500;//連接超時時間(ms)

        //錯誤的token值連接示例
//        final Socket socket = IO.socket("http://localhost:9092/?token=123456", options);

        //常規連接
//        final Socket socket = IO.socket("http://localhost:9092/?token=87df42a424c48313ef6063e6a5c63297", options);

        //連接到指定的聊天室
        final Socket socket = IO.socket("http://localhost:9092/room2?token=87df42a424c48313ef6063e6a5c63297", options);

        socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                socket.send("hello");
            }
        });

        socket.on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                System.out.println("連接關閉");
            }
        });

        socket.on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
            @Override
            public void call(Object... args) {
                System.out.println("sessionId:" + socket.id());
                for (Object obj : args) {
                    System.out.println(obj);
                }
                System.out.println("收到服務器應答，將要斷開連接...");
                socket.disconnect();
            }
        });
        socket.connect();
    }
}

注意上面連接時，room1的指定，其它就不多說了，代碼就是最好的注釋:)