socket.io 在java与微信小程序上的应用

ゝ一纸荒年。 2022-04-18 06:12 1026阅读 0赞

最近有一个这样的功能场景。用户操作完成后。服务器主动通知另一个客户端显示结果。

这里涉及一个服务器推的这么一个东西。需要实现这么一个功能，对比了几个实现方式。最终选择了socket.io。

1、commet，最初想到这个功能要求不高，想简单的通过commet方式实现就算了。但考虑到commet已经是比较老旧的做法了，其中有很多弊端，所以放弃了。

2、netty。netty是非常棒的java nio框架。但考虑到我需要实现的功能，有点杀鸡用牛刀的感觉，而且要兼顾小程序的实现，选择放弃

3、socket.io，非常适合。封装了socket的特性，用法简单。而且对不同语言的兼容性很好。

废话好说。直接看代码。

1、服务器端，使用的是java语言。

首先添加jar支持，java上服务器端用的是netty-socketio这个jar，是对netty做的二次封装。

<dependency>
    			<groupId>com.corundumstudio.socketio</groupId>
    			<artifactId>netty-socketio</artifactId>
    			<version>1.7.16</version>
    		</dependency>

java服务器端的实现：

/*
     * 文件名：SocketIoServer.java 版权：Copyright by www.poly.com 描述： 修改人：gogym 修改时间：2018年10月23日 跟踪单号： 修改单号：
     * 修改内容：
     */
    
    
    import java.util.HashMap;
    import java.util.Map;
    
    import com.corundumstudio.socketio.AckCallback;
    import com.corundumstudio.socketio.AckRequest;
    import com.corundumstudio.socketio.AuthorizationListener;
    import com.corundumstudio.socketio.Configuration;
    import com.corundumstudio.socketio.HandshakeData;
    import com.corundumstudio.socketio.SocketIOClient;
    import com.corundumstudio.socketio.SocketIONamespace;
    import com.corundumstudio.socketio.SocketIOServer;
    import com.corundumstudio.socketio.listener.ConnectListener;
    import com.corundumstudio.socketio.listener.DataListener;
    import com.corundumstudio.socketio.listener.DisconnectListener;
    import com.corundumstudio.socketio.listener.PingListener;
    
    
    public class SocketIoServer
    {
    
        public static void main(String[] args)
        {
    
            // 启动服务
            SocketIoServer sio = new SocketIoServer();
            sio.initSocket();
    
        }
    
        // 保存session与端映射关系
        public Map<String, SocketIOClient> clientMap = new HashMap<String, SocketIOClient>();
    
        /**
         * Description: 初始化
         * 
         * @see
         */
        public void initSocket()
        {
    
            Configuration config = new Configuration();
            // 一般不需要设置Hostname，设置localhost后，从其他IP连接会连不上来
            // config.setHostname("localhost");
            // 设置端口
            config.setPort(10015);
            // 往服务器写ping消息，用来检测客户端是否存活，如果设置时间内没有ping通，则清理掉客户端连接
            config.setPingTimeout(3000);
            // ping间隔时长
            config.setPingInterval(30000);
            // config.setBossThreads(5);
            // config.setWorkerThreads(15);
    
            config.setAuthorizationListener(new AuthorizationListener()
            {
                @Override
                public boolean isAuthorized(HandshakeData handshakedata)
                {
                    // 这里可以拦截连接过来的URL。可以获取参数等。比如统一鉴权就可以在这里处理
                    System.out.println(handshakedata.getUrl());
                    return true;
                }
            });
    
            SocketIOServer server = new SocketIOServer(config);
    
            // 添加一个命名空间，命名空间可以用于区分不同的业务连接
            SocketIONamespace namespace = server.addNamespace("/chat");
    
            server.start();
        }
    
        /**
         * Description: 添加监听程序
         * 
         * @param namespace
         * @see
         */
        public void addListener(SocketIONamespace namespace)
        {
    
            // 添加连接监听
            namespace.addConnectListener(new ConnectListener()
            {
                @Override
                public void onConnect(SocketIOClient client)
                {
                    System.out.println(client.getRemoteAddress() + "连接上了");
                    // 把连接上的客户端保存起来
                    clientMap.put(client.getSessionId().toString(), client);
    
                }
            });
    
            // 添加连接断开监听
            namespace.addDisconnectListener(new DisconnectListener()
            {
                @Override
                public void onDisconnect(SocketIOClient client)
                {
                    System.out.println(client.getRemoteAddress() + "离开了");
                    clientMap.remove(client.getSessionId().toString());
                }
            });
    
            // 添加ping监听
            namespace.addPingListener(new PingListener()
            {
                @Override
                public void onPing(SocketIOClient socketioclient)
                {
                    // 往客户端写ping消息，用来检测客户端是否存活
                    System.out.println("ping:" + socketioclient.getRemoteAddress());
                }
            });
    
            // 添加自定义监听，例如：用户注册
            namespace.addEventListener("register_event", String.class, new DataListener<String>()
            {
    
                @Override
                public void onData(SocketIOClient client, String str, AckRequest ack)
                    throws Exception
                {
                    // 发送ack给客户端告知服务器端已经收到消息
                    ack.sendAckData("ok");
                }
            });
    
        }
    
        public void sendMsg(String sessionId, String msg)
        {
    
            // 通过sessionid获取socket连接
            SocketIOClient client = clientMap.get(sessionId);
    
            // 发条消息给客户端。客户端监听的是message
            client.sendEvent("message", new AckCallback<Object>(Object.class, 5 /* 注意这里的单位是秒 */)
            {
                @Override
                public void onSuccess(Object result)
                {
                    // 接收客户端返回的ack。意味着客户端已经接收到消息了
                    System.out.println("ack from client: " + client.getSessionId() + " data: "
                                       + result);
                }
    
                @Override
                public void onTimeout()
                {
                    System.out.println("ACK超时");
                }
    
            }, msg);
    
        }
    
    }

JAVA客户端的实现

/*
     * 文件名：SocketIoClient.java 版权：Copyright by www.poly.com 描述： 修改人：gogym 修改时间：2018年10月23日 跟踪单号： 修改单号：
     * 修改内容：
     */
    
    package com.poly.rbl.plugin.socketio;
    
    
    import io.socket.client.Ack;
    import io.socket.client.IO;
    import io.socket.client.Socket;
    import io.socket.emitter.Emitter;
    
    import com.poly.rbl.utils.DateTimeUtil;
    
    
    public class SocketIoClient
    {
    
        public static void main(String[] args)
        {
    
            IO.Options options = new IO.Options();
            // 设置协议为websocket
            options.transports = new String[] {"websocket"};
            // 失败重连次数
            options.reconnectionAttempts = 5;
            // 失败重连的时间间隔
            options.reconnectionDelay = 3000;
            // 连接超时时间(ms)
            options.timeout = 3000;
            // 开启重连
            options.reconnection = true;
            // 可以携带一些连接参数
            // options.query="token=1234";
    
            try
            {
                final Socket socket = IO.socket("http://app.52rbl.com:8004/delivery", options);
                // 连接服务器
                socket.on(Socket.EVENT_CONNECT, new Emitter.Listener()
                {
                    @Override
                    public void call(Object... args)
                    {
                        System.out.println(DateTimeUtil.getCurrentTime() + ":client connect! ");
    
                        // 连接成功，马上向服务器发送信息，例如：用户注册
                        socket.emit("register_event", "发给服务器的消息", new Ack()
                        {
                            @Override
                            public void call(Object... aobj)
                            {
                                System.out.println(aobj[0]);
                            }
                        });
                    }
                });
    
                // 监听断线
                socket.on(Socket.EVENT_DISCONNECT, new Emitter.Listener()
                {
                    @Override
                    public void call(Object... args)
                    {
                        System.out.println(DateTimeUtil.getCurrentTime() + ":client disconnect!");
                    }
                });
    
                // 监听ping,其实这个监听没有多少意义
                socket.on(Socket.EVENT_PING, new Emitter.Listener()
                {
                    @Override
                    public void call(Object... arg0)
                    {
                        // 往服务器写ping
                        System.out.println("ping:" + arg0);
                    }
                });
    
                // 监听pong,这个监听没有多少意义
                socket.on(Socket.EVENT_PONG, new Emitter.Listener()
                {
                    @Override
                    public void call(Object... arg0)
                    {
                        // ping了以后接收服务器的pong响应
                        System.out.println("pong:" + arg0[0]);
                    }
                });
    
                // 监听服务器发送的消息，也可以自定义一个简单字符串
                socket.on(Socket.EVENT_MESSAGE, new Emitter.Listener()
                {
                    @Override
                    public void call(Object... args)
                    {
                        // 接收消息后，往服务器发送一个ack,告诉服务器消息收到了
                        Ack ack = (Ack)args[args.length - 1];
                        ack.call(args[0].hashCode());
    
                        for (Object obj : args)
                        {
                            System.out.println(DateTimeUtil.getCurrentTime()
                                               + ":receive server message=" + obj.toString());
                        }
                    }
                });
    
                // 连接服务器
                socket.connect();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    
    }

小程序上也可以使用，虽然小程序有提供websocket的支持，但并不支持socket.io。如果想要在小程序上使用，需要添加一个小程序上的专用库：[weapp.socket.io][]    [https://github.com/10cella/weapp.socket.io][https_github.com_10cella_weapp.socket.io]

完

[weapp.socket.io]: http://weappsocket.matong.io/
[https_github.com_10cella_weapp.socket.io]: https://github.com/10cella/weapp.socket.io