远方蔚蓝
一刹那情真，相逢不如不见

文章数量 126

访问次数 199896

运行天数 1437

最近活跃 2024-10-04 23:36:48

进入后台管理系统

博客 - 不使用web容器搭建一个简单的websocket服务器

不使用web容器搭建一个简单的websocket服务器

  1. 由于使用注解@ServerEndpoint来搭建Websocket服务，不知道怎么指定端口
  1. 使用注解假设不使用容器（tomcat、jetty）的话，实在是不知道怎么启动服务端，另外使用注解启动服务器还需要在@ServerEndpoint注解里指定一个访问路径如@ServerEndpoint("路径")，访问时需要ws://IP:端口/项目名/路径这样访问，如果不这样访问就不知道还有没有其他方法
  1. 假设就使用默认80端口，接下来我还是不知道如何启动它。
  1. 使用下面这种方法是因为使用继承WebSocketServer类时，可以使用封装的start或者run方法来启动，这些都是现成的。
slf4j-log4j12-1.7.7
slf4j-api-1.7.22
slf4j-api-1.7.7
log4j-1.2.17
Java-WebSocket-1.5.1
fastjson-1.2.4
log4j.rootLogger=info, stdout,E, file
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %c{1}:%L - %m%n
​
log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
log4j.appender.file.File=logs/runner.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd|HH\:mm\:ss} %5p %c{1}\:%L - %m%n
​
log4j.appender.E=org.apache.log4j.DailyRollingFileAppender
log4j.appender.E.File=logs/error.log
log4j.appender.E.Threshold=ERROR
log4j.appender.E.layout=org.apache.log4j.PatternLayout
log4j.appender.E.layout.ConversionPattern=%-d{yyyy-MM-dd HH\:mm\:ss} [%p]-%m%n
  • 发送信息的客户端
package test;
​
import java.net.URI;
import java.util.concurrent.TimeUnit;
​
import org.java_websocket.enums.ReadyState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
​
import socket.SyncWebSocketClient;
​
/**
 * WebSocketClient
 * @author wst 2022年11月29日 下午3:40:52
 *
 */
public class TestWebSocketClient {
​
	private static SyncWebSocketClient client;
	private static Logger logger = LoggerFactory.getLogger(TestWebSocketClient.class);
	
	private static void connectOtherWebsocketServer(String addr) throws Exception {
 		try {
			if (client == null || client.getReadyState().equals(ReadyState.NOT_YET_CONNECTED)) {
			    client = new SyncWebSocketClient(new URI(addr));
			    boolean flag = client.connectBlocking(60, TimeUnit.SECONDS);
			    if (!flag) {
			        client = null;
			        logger.info("SyncWebSocketClient连接失败");
			    } else {
			    	logger.info("SyncWebSocketClient连接成功");
			    }
			} else if (client.getReadyState().equals(ReadyState.CLOSING) || client.getReadyState().equals(ReadyState.CLOSED)) {
				logger.info("SyncWebSocketClient连接已关闭，准备重连");
			    client.reconnect();
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.info("[异常]连接服务异常," + e.getMessage());
			throw new Exception("连接服务异常," + e.getMessage());
		}
 	}
	
	static int i = 0;
	public static void main(String[] args) throws Exception {
		connectOtherWebsocketServer("ws://127.0.0.1:8095");
		client.setConnectionLostTimeout(0);
		System.out.println(client.sendAndGet("888888"));
	}
}
​
import java.net.URI;
import java.util.concurrent.CountDownLatch;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
/**
 * WebSocket 同步客户端
 */
public class SyncWebSocketClient extends WebSocketClient {
    private String         message;
    private Object         lock      = new Object();
    private CountDownLatch downLatch = null;
    public SyncWebSocketClient(URI serverUri) {
        super(serverUri);
        // this.waitConnect();
    }
​
    public void waitConnect() {
        try {
            this.connectBlocking();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
​
    @Override
    public void onOpen(ServerHandshake serverHandshake) {
​
    }
​
    @Override
    public void onMessage(String msg) {
        message = msg;
        if (downLatch != null) {
            //释放锁资源
            downLatch.countDown();
        }
    }
​
    @Override
    public void onClose(int i, String s, boolean b) {
        if (downLatch != null) {
            // 释放锁资源
            downLatch.countDown();
        }
    }
​
    @Override
    public void onError(Exception e) {
        // 释放锁资源
        if (downLatch != null) {
            downLatch.countDown();
        }
    }
​
    public String sendAndGet(String text) throws Exception {
        synchronized (lock) {
            message = null;
            downLatch = new CountDownLatch(1);
            this.send(text);
            downLatch.await();
        }
        return message;
    }
    
    @Override
    public void onWebsocketPing(WebSocket conn, Framedata f) {
    	super.onWebsocketPing(conn, f);
    }
    
    @Override
    public void onWebsocketPong(WebSocket conn, Framedata f) {
    	super.onWebsocketPong(conn, f);
    }
​
    public String getMessage() {
        return message;
    }
​
    public void setMessage(String message) {
        this.message = message;
    }
}
​
  • 服务器一
package test;
​
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
​
import org.java_websocket.WebSocket;
import org.java_websocket.enums.ReadyState;
import org.java_websocket.framing.Framedata;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
​
import socket.SyncWebSocketClient;
​
/**
 * WebSocketServer服务器一
 * @author wst(shi-tao.wen) 2022年2月22日 下午1:54:41
 *
 */
public class TestWebSocketServer1 extends WebSocketServer{
	
	private static Logger logger = LoggerFactory.getLogger(TestWebSocketServer1.class);
	private static SyncWebSocketClient client;
​
    /**
     * 存放接入的socket连接
     */
    private static Map<String, WebSocket> clientMap = new HashMap<String, WebSocket>();
​
    public TestWebSocketServer1(int port) throws UnknownHostException{
    	super(new InetSocketAddress(InetAddress.getByName("0.0.0.0"), port));
    	logger.info("构建websocket服务器 {}:{}", InetAddress.getByName("0.0.0.0"), port);
    }
    
    public TestWebSocketServer1() {
    	
    }
​
    public TestWebSocketServer1(InetSocketAddress address) {
        super(address);
    }
​
    /**
     * 当服务器启动时
     */
    @Override
	public void onStart() {
    	logger.info("WebSocketSelfServer onStart() : {}", "");
	}
    
    @Override
    public void onOpen(WebSocket webSocket, ClientHandshake clientHandshake) {
    	logger.info("客户端[{}]接入!", webSocket.getRemoteSocketAddress());
    	
    	// 保存接入的socket连接
    	clientMap.put(webSocket.getRemoteSocketAddress().toString(), webSocket);
    	
    	try {
			connectOtherWebsocketServer("ws://127.0.0.1:8096");
		} catch (Exception e) {
			e.printStackTrace();
		}
    }
​
    /**
     * 当接收到信息时
     */
    @Override
    public void onMessage(WebSocket webSocket, String message) {
    	// 这里开个线程，因为client阻塞的问题
    	new Thread(new Runnable() {
			
			@Override
			public void run() {
				logger.info("客户端[{}]收到：{}", webSocket.getRemoteSocketAddress(), message);
		    	try {
					webSocket.send(client.sendAndGet(message));
				} catch (Exception e1) {
					e1.printStackTrace();
				}				
			}
		}).start();
    }
​
    /**
     * 当连接断开时
     */
    @Override
    public void onClose(WebSocket webSocket, int n, String str, boolean bool) {
    	logger.info("客户端[{}]断开!", webSocket.getRemoteSocketAddress());
    	
    	// 移除断开的socket
    	clientMap.remove(webSocket.getRemoteSocketAddress().toString());
    }
    
    @Override
    public void onError(WebSocket webSocket, Exception e) {
    	logger.info("[异常]WebSocketSelfServer onError()");
        e.printStackTrace();
    }
    
	/**
	 * 接收到客户端发来的ping
	 */
	@Override
	public void onWebsocketPing(WebSocket webSocket, Framedata f) {
		logger.info("onWebsocketPing");
		super.onWebsocketPing(webSocket, f);
	}
	
	/**
	 * 接收到客户端发来的pong
	 */
	@Override
	public void onWebsocketPong(WebSocket webSocket, Framedata f) {
		logger.info("onWebsocketPong");
		super.onWebsocketPong(webSocket, f);
	}	
	
	public static void main(String[] args) throws UnknownHostException {
		new TestWebSocketServer1(8095).start();
	}
	
	private static void connectOtherWebsocketServer(String addr) throws Exception {
 		try {
			if (client == null || client.getReadyState().equals(ReadyState.NOT_YET_CONNECTED)) {
			    client = new SyncWebSocketClient(new URI(addr));
			    boolean flag = client.connectBlocking(60, TimeUnit.SECONDS);
			    if (!flag) {
			        client = null;
			        logger.info("SyncWebSocketClient连接失败");
			    } else {
			    	logger.info("SyncWebSocketClient连接成功");
			    }
			} else if (client.getReadyState().equals(ReadyState.CLOSING) || client.getReadyState().equals(ReadyState.CLOSED)) {
				logger.info("SyncWebSocketClient连接已关闭，准备重连");
			    client.reconnect();
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.info("[异常]连接服务异常," + e.getMessage());
			throw new Exception("连接服务异常," + e.getMessage());
		}
 	}
}
​
  • 服务器二
package test;
​
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
​
import org.java_websocket.WebSocket;
import org.java_websocket.framing.Framedata;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
​
/**
 * WebSocketServer服务器二
 * @author wst(shi-tao.wen) 2022年2月22日 下午1:54:41
 *
 */
public class TestWebSocketServer2 extends WebSocketServer{
	
	private Logger logger = LoggerFactory.getLogger(this.getClass());
​
    /**
     * 存放接入的socket连接
     */
    private static Map<String, WebSocket> clientMap = new HashMap<String, WebSocket>();
​
    public TestWebSocketServer2(int port) throws UnknownHostException{
    	super(new InetSocketAddress(InetAddress.getByName("0.0.0.0"), port));
    	logger.info("构建websocket服务器 {}:{}", InetAddress.getByName("0.0.0.0"), port);
    }
    
    public TestWebSocketServer2() {
    	
    }
​
    public TestWebSocketServer2(InetSocketAddress address) {
        super(address);
    }
​
    /**
     * 当服务器启动时
     */
    @Override
	public void onStart() {
    	logger.info("WebSocketSelfServer onStart() : {}", "");
	}
    
    @Override
    public void onOpen(WebSocket webSocket, ClientHandshake clientHandshake) {
    	logger.info("客户端[{}]接入!", webSocket.getRemoteSocketAddress());
    	
    	// 保存接入的socket连接
    	clientMap.put(webSocket.getRemoteSocketAddress().toString(), webSocket);
    }
​
    /**
     * 当接收到信息时
     */
    @Override
    public void onMessage(WebSocket webSocket, String message) {
    	logger.info("客户端[{}]收到：{}", webSocket.getRemoteSocketAddress(), message);
    	webSocket.send("-------------");
    }
​
    /**
     * 当连接断开时
     */
    @Override
    public void onClose(WebSocket webSocket, int n, String str, boolean bool) {
    	logger.info("客户端[{}]断开!", webSocket.getRemoteSocketAddress());
    	
    	// 移除断开的socket
    	clientMap.remove(webSocket.getRemoteSocketAddress().toString());
    }
    
    @Override
    public void onError(WebSocket webSocket, Exception e) {
    	logger.info("[异常]WebSocketSelfServer onError()");
        e.printStackTrace();
    }
    
	/**
	 * 接收到客户端发来的ping
	 */
	@Override
	public void onWebsocketPing(WebSocket webSocket, Framedata f) {
		logger.info("onWebsocketPing");
		super.onWebsocketPing(webSocket, f);
	}
	
	/**
	 * 接收到客户端发来的pong
	 */
	@Override
	public void onWebsocketPong(WebSocket webSocket, Framedata f) {
		logger.info("onWebsocketPong");
		super.onWebsocketPong(webSocket, f);
	}	
	
	public static void main(String[] args) throws UnknownHostException {
		new TestWebSocketServer2(8096).start();
	}
}
​

给我留言

评论