专栏首页菩提树下的杨过netty-socketio 示例代码

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的指定,其它就不多说了,代码就是最好的注释:)

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Spring Security笔记:Remember Me(下次自动登录)

    前一节学习了如何限制登录尝试次数,今天在这个基础上再增加一点新功能:Remember Me. 很多网站,比如博客园,在登录页面就有这个选项,勾选“下次自动登录”...

    菩提树下的杨过
  • Flash/Flex学习笔记(30):不用startDrag和stopDrag的对象拖动

    对于从Sprite类继承来的对象,要实现拖放当然是Flash/Flex学习笔记(13):对象拖动(startDrag/stopDrag) 里讲的方法最方便,但是...

    菩提树下的杨过
  • scala 学习笔记(07) 一等公民的函数

    在scala中一切皆对象,一切皆函数,函数跟Int,String、Class等其它类型是处于同等的地位,换句话说,使用函数跟使用普通的类型一样,没什么区别,因此...

    菩提树下的杨过
  • web项目对接钉钉扫码登录

    今天我们记录一下关于vue进行web开发的过程中对接钉钉的H5微应用的时候扫码登录的功能,你说他难吧,其实不难,很简单,你说他简单吧,看文档可能真的有点乱,不然...

    何处锦绣不灰堆
  • Spring Security笔记:Remember Me(下次自动登录)

    前一节学习了如何限制登录尝试次数,今天在这个基础上再增加一点新功能:Remember Me. 很多网站,比如博客园,在登录页面就有这个选项,勾选“下次自动登录”...

    菩提树下的杨过
  • 玩了一下websocket 原

    domain0
  • 无回路有向图的拓扑排序

    因公司业务需要,在表单中每个字段都会配置自动计算,但自动计算公式中会引用到其他字段中的值。所以希望可以根据计算公式,优先计算引用的公式。所以最终使用了无回路有向...

    兜兜毛毛
  • MHA搭建之ssh互信打通脚本

    在MySQL搭建MHA高可用架构的时候,需要打通master、slave、以及mha manager之间的ssh互信,通常情况下,运维人员需要手动打通ssh互信...

    AsiaYe
  • CSS实现兼容性的渐变背景(gradient)效果兼容众多浏览器

    落叶大大
  • 设计模式 | 策略模式

    定义:将一组特定的行为和算法封装成类,以适应某些特定的上下文环境,并让它们可以相互替换,这种模式就是策略模式。

    憧憬博客

扫码关注云+社区

领取腾讯云代金券