java-websocket客户端断线重连注入Service问题

java版客户端：

使用开源项目java-websocket， github地址： https://github.com/TooTallNate/Java-WebSocket

github上有很多示例，具体可以去查看

此处主要是记录java-websocket实现客户端，并解决无法使用Service层方法（service为null）的问题，以及断线重连

引用包

<dependency>
    <groupId>org.java-websocket</groupId>
    <artifactId>Java-WebSocket</artifactId>
    <version>1.3.9</version>
</dependency>

第一版，使用getBean获取Service层方法，并且实现断线重连

使用的是GitHub上的demo示例

import com.alibaba.fastjson.JSONArray;import com.sensor.vibration.utils.ApplicationContextRegister;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

import java.net.URI;
import java.util.Map;

/** This example demonstrates how to create a websocket connection to a server. Only the most important callbacks are overloaded. */

@Slf4j
public class SensorWebSocketClient extends WebSocketClient {
    @Autowired
    private UserService userService;

    public SensorWebSocketClient(URI serverUri , Draft draft ) {
        super( serverUri, draft );
    }

    public SensorWebSocketClient(URI serverURI ) {
        super( serverURI );
    }

    public SensorWebSocketClient(URI serverUri, Map<String, String> httpHeaders ) {
        super(serverUri, httpHeaders);
    }

    @Override
    public void onOpen( ServerHandshake handshakedata ) {
        System.out.println( "opened connection" );
        // if you plan to refuse connection based on ip or httpfields overload: onWebsocketHandshakeReceivedAsClient
    }

    @Override
    public void onMessage( String msg ) {
        log.info("[websocket] 收到消息={}",msg);
    }

    @Override
    public void onClose( int code, String reason, boolean remote ) {
        // The codecodes are documented in class org.java_websocket.framing.CloseFrame
        System.out.println( "Connection closed by " + ( remote ? "remote peer" : "us" ) + " Code: " + code + " Reason: " + reason );
    }

    @Override
    public void onError( Exception ex ) {
        ex.printStackTrace();
        // if the error is fatal then onClose will be called additionally
    }

}

新建一个类，创建一个方法，启动websocket

import java.net.URI;
import java.net.URISyntaxException;


/**
 * Simple example to reconnect blocking and non-blocking.
 */

public class ReconnectClient {
    public static void reconnect() throws URISyntaxException, InterruptedException{
        SensorWebSocketClient c = new SensorWebSocketClient( new URI( "ws://localhost:5005/websocket" ) );
        c.connectBlocking();

        new Thread(new Runnable() {
            public void run() {
                System.out.println("Runnable running..");
            }
        }) {
            public void run() {
                while (true){
                    try{
                        Thread.sleep(3000);
                        c.send("");
                    }catch (Exception e){
                        c.reconnect();
                    }
                }
            };
        }.start();
    }
}

在新建一个类，程序启动的时候，调用上面的方法

import com.sensor.vibration.utils.Common;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.URISyntaxException;
import java.util.Map;


@Slf4j
@Component
public class InitStart implements CommandLineRunner {

    @Override
    public void run(String... args) throws URISyntaxException, InterruptedException{
        ReconnectClient.reconnect();
    }

}

中间的启动类的方法可以省去，直接写在InitStart的run方法里面

现在还不能使用Service层的方法，会报service为null异常，百度后，参考别人使用getBean方法，写一个工具类

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;


@Component
@Lazy(false)
public class ApplicationContextRegister  implements ApplicationContextAware {
    private static ApplicationContext APPLICATION_CONTEXT;

    /**
     * 设置spring上下文  *  * @param applicationContext spring上下文  * @throws BeansException
     */

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        APPLICATION_CONTEXT = applicationContext;
    }

    public static ApplicationContext getApplicationContext() {
        return APPLICATION_CONTEXT;
    }


}

在 SensorWebSocketClient.java 中使用Service

@Autowired
private UserService userService;

ApplicationContext act = ApplicationContextRegister.getApplicationContext();
userService=act.getBean(UserService.class);

但是领导不让用getBean这种方法，放弃

第二版，使用Service层方法版本 + 断线重连

实现：

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.URI;

/**
 * Created by Chow on 2019/8/22
 */

@Slf4j
@Component
public class WebSocketClientStart {
    @Autowired
    private UserService userService;


    @Bean
    public WebSocketClient webSocketClient() {
        try {
            WebSocketClient webSocketClient = new WebSocketClient(new URI("ws://127.0.0.1:5005/websocket")) {
                @Override
                public void onOpen(ServerHandshake handshakedata) {
                    log.info("[websocket] 连接成功");
                }

                @Override
                public void onMessage(String msg) {
                    try{
                        log.info("[websocket] 收到消息={}",msg);

                        if (msg == null || StringUtils.isBlank(msg)){
                            log.error("the msg message of websocket received is null");
                            this.send("");
                            return;
                        }
                        JSONArray jsonArray = JSONArray.parseArray(msg);
                        if (jsonArray == null || jsonArray.size() == 0){
                            log.info("log: the message of websocket received is empty");
                        }

                        vibrationAlarmService.alarmAnalysis(jsonArray);
                        this.send("");
                    }catch (Exception e){
                        log.error(e.getMessage(), e);
                        this.send("");
                    }

                }

                @Override
                public void onClose(int code, String reason, boolean remote) {
                    log.info("[websocket] 退出连接");
                }

                @Override
                public void onError(Exception ex) {
                    log.info("[websocket] 连接错误={}",ex.getMessage());
                }
            };
            webSocketClient.connect();
            new Thread(new Runnable() {

                public void run() {
                    System.out.println("Runnable running..");
                }

            }) {

                public void run() {
                    while (true){
                        try{
                            Thread.sleep(3000);
                            webSocketClient.send("");
                        }catch (Exception e){
                            webSocketClient.reconnect();
                        }
                    }

                }

            }.start();
            return webSocketClient;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


}

测试服务端代码

需要引入包

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
    <version>2.1.3.RELEASE</version>
</dependency>

server：

import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;

import org.springframework.stereotype.Component;

@ServerEndpoint(value = "/websocket")
@Component
public class MyWebSocket {
    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    /**
     * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
        try {
            sendMessage("当前在线人数为" + getOnlineCount());
        } catch (IOException e) {
            System.out.println("IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        System.out.println("来自客户端的消息:" + message);

        //群发消息
        for (MyWebSocket item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @OnError
    public void onError(Session session, Throwable error) {
        System.out.println("发生错误");
        error.printStackTrace();
    }


    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 群发自定义消息
     * */
    public static void sendInfo(String message) throws IOException {
        for (MyWebSocket item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                continue;
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        MyWebSocket.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        MyWebSocket.onlineCount--;
    }
}

java-websocket客户端 断线重连 注入Service问题

第一版，使用getBean获取Service层方法，并且实现断线重连

第二版，使用Service层方法版本 + 断线重连

测试服务端代码

java-websocket客户端断线重连注入Service问题