是的 本文介绍java如何作为客户端 发起websocket请求
博主不做标题党 不会服务端客户端分不清就写个标题 乱写文章
前言
为什么会使用java作为websocket客户端? 虽说websocket协议 本意是web与服务端之间的通讯协议,那假设有一天 我们的供应商 或者是甲方大爷 只提供了websocket接口呢?
如果直接让前端去对接,再把数据传给后端,那从前端对接到后端入库的步骤,万一出现数据丢失呢? 总之把数据处理放在后端,是相对可靠的, 我们可以借助netty来实现websocket客户端功能
长链接有两点值得注意的,一是心跳机制 二是重连机制
如果不发送心跳包,可能过会儿连接就断开了;
重新机制就比较好理解了,不管是服务端还是客户端的断开,作为客户端都需要能够重连
代码
博主对长链接并不是特别熟练,但是什么代码是能正式用的,什么代码上不了生产只是个写着玩的demo 还是一眼能分辨出来的,代码主要参考 git@github.com:yimiancheng/netty-study.git
,写的很不错,代码优化空间很少了, 线程池博主是推荐用new ThreadPoolExecutor()创建,避免OOM的问题,除此之外应该是准生产环境级别代码了。
maven依赖
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.35.Final</version>
</dependency>
websocket客户端handle类 主要处理接收的消息、事件等,
事件触发会进入userEventTriggered方法
建立连接会进入channelActive方法
接收消息会进入channelRead0方法
出现异常会进入exceptionCaught方法
断开连接会进入channelInactive方法(本文中未重写该方法, 在channelInactive方法中重连也是没问题的)
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.SocketClient;
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.websocket.SendMsg;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* WebSocketClientFrameHandler
*
* @date 2019/8/20 16:42
*/
public class WebSocketClientFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
private static final Logger LOG = LoggerFactory.getLogger(WebSocketClientFrameHandler.class);
private SocketClient socketClient;
private ChannelPromise channelPromise;
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
LOG.info("客户端接收到事件 " + (evt.getClass()) + " | " + evt.toString());
if(WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE.equals(evt)) {
LOG.info(ctx.channel().id().asShortText() + " 握手完成!");
socketClient.CHANNEL_IS_READY.set(true);
channelPromise.setSuccess();
// SendMsg.send(ctx.channel(),"客户端握手完成消息 -》服务器时间: " + System.currentTimeMillis());
}
else if(evt instanceof IdleStateEvent){
//ctx.channel().writeAndFlush(new PingWebSocketFrame());
IdleStateEvent evtIdle = (IdleStateEvent) evt;
switch(evtIdle.state()) {
case WRITER_IDLE:
// SendMsg.send(ctx.channel(),"客户端 ping 消息 -》服务器时间: " + System.currentTimeMillis());
ctx.channel().writeAndFlush(new PingWebSocketFrame());
case READER_IDLE:
// SendMsg.send(ctx.channel(),"客户端 ping 消息 -》服务器时间: " + System.currentTimeMillis());
ctx.channel().writeAndFlush(new PingWebSocketFrame());
default:
break;
}
}
super.userEventTriggered(ctx, evt);
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
LOG.info("打开连接 handlerAdded SUCCESS. | name = " +channel.id().asShortText());
super.handlerAdded(ctx);
channelPromise = ctx.newPromise();
}
/**
* Channel 已经被注册到了EventLoop
*/
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
LOG.info("注册成功 channelRegistered SUCCESS. | name = " +channel.id().asShortText());
super.channelRegistered(ctx);
}
/**
* Channel 处于活动状态(已经连接到它的远程节点)。它现在可以接收和发送数据了
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
LOG.info("活动状态 channelActive SUCCESS. | name = " +channel.id().asShortText());
super.channelActive(ctx);
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame) throws
Exception {
if(webSocketFrame instanceof TextWebSocketFrame) {
// String message = textWebSocketFrame.content().toString(Charset.forName("utf-8"));
String message = ((TextWebSocketFrame) webSocketFrame).text();
LOG.info("客户端接收到消息:" + message);
}
else {
LOG.info("接收到消息类型" + (webSocketFrame.getClass().getName()));
}
SendMsg.sendPong(channelHandlerContext.channel());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
LOG.error("消息处理失败: " + cause.getMessage(), cause);
ctx.close();
}
/**
* Channel 已经被创建,但还未注册到EventLoop
* 连接断开
*/
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
LOG.info("连接断开 channelUnregistered SUCCESS. | name = " +channel.id().asShortText());
super.channelUnregistered(ctx);
channelPromise = null;
}
public SocketClient getSocketClient() {
return socketClient;
}
public void setSocketClient(SocketClient socketClient) {
this.socketClient = socketClient;
}
public ChannelPromise getChannelPromise() {
return channelPromise;
}
}
心跳任务类
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.SocketClient;
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.websocket.SendMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.TimerTask;
/**
* HeartBeatTimerTask
*
* @date 2019/9/2 16:24
*/
public class HeartBeatTimerTask extends TimerTask {
private static final Logger LOG = LoggerFactory.getLogger(ReconnectTimerTask.class);
private SocketClient socketClient;
public HeartBeatTimerTask(SocketClient socketClient) {
this.socketClient = socketClient;
}
@Override
public void run() {
if(socketClient != null && socketClient.isValid()) {
SendMsg.send(socketClient.getChannel(), "客户端心跳消息 => " + System.currentTimeMillis());
}
}
}
重连任务类
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.SocketClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.TimerTask;
/**
* ReconnectTimerTask
*
* @date 2019/9/2 16:03
*/
public class ReconnectTimerTask extends TimerTask {
private static final Logger LOG = LoggerFactory.getLogger(ReconnectTimerTask.class);
private SocketClient socketClient;
public ReconnectTimerTask(SocketClient socketClient) {
this.socketClient = socketClient;
}
@Override
public void run() {
if(socketClient != null && !socketClient.isValid()) {
LOG.info("=== 客户端重连 " + System.currentTimeMillis());
socketClient.connect();
}
}
}
自定义线程工厂类,主要是给线程重命名 便于维护、调试
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
public class NamedThreadFactory implements ThreadFactory {
static final AtomicInteger poolNumber = new AtomicInteger(1);
final AtomicInteger threadNumber = new AtomicInteger(1);
final ThreadGroup group;
final String prefix;
final boolean isDaemon;
final int priority;
public NamedThreadFactory() {
this("pool");
}
public NamedThreadFactory(String prefix) {
this(prefix, false, Thread.NORM_PRIORITY);
}
public NamedThreadFactory(String prefix, boolean isDaemon, int priority) {
SecurityManager sm = System.getSecurityManager();
this.group = (sm != null) ? sm.getThreadGroup() : Thread.currentThread().getThreadGroup();
this.prefix = prefix + "-" + poolNumber.getAndIncrement() + "-thread-";
this.isDaemon = isDaemon;
this.priority = priority;
}
public Thread newThread(Runnable r) {
Thread thread = new Thread(group, r, prefix + threadNumber.getAndIncrement(), 0);
thread.setDaemon(isDaemon);
thread.setPriority(priority);
return thread;
}
}
消息监听类,主要用于发送消息(包括心跳包)是否成功
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
class CustomerChannelFutureListener implements ChannelFutureListener {
private static final Logger LOG = LoggerFactory.getLogger(CustomerChannelFutureListener.class);
public void operationComplete(ChannelFuture channelFuture) throws Exception {
// LOG.info(JSON.toJSONString(channelFuture));
if(channelFuture.isDone() && channelFuture.isSuccess()){
// LOG.info("send success.");
}
else {
channelFuture.channel().close();
LOG.info("send error. cause = " + channelFuture.cause());
channelFuture.cause().printStackTrace();
}
}
}
简单封装的消息发送类
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.commons.collections.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* SendMsg
*
* @date 2019/8/30 18:02
*/
public class SendMsg {
private static final Logger LOG = LoggerFactory.getLogger(SendMsg.class);
public static ConcurrentHashMap<String, Channel> ALL_CHANNEL = new ConcurrentHashMap<String, Channel>();
public static void startSendMsg() {
Thread thread = new Thread(new Runnable() {
public void run() {
while(true) {
sendMsgTest();
}
}
});
thread.start();
}
public static void sendMsgTest() {
try {
Map<String, Channel> map = Collections.unmodifiableMap(ALL_CHANNEL);
LOG.info("map size = " + map.size());
if(MapUtils.isEmpty(map)) {
Thread.sleep(2000);
return;
}
for(Map.Entry<String, Channel> entry : map.entrySet()) {
LOG.info("------------- key = " + entry.getKey());
send(entry.getValue(), "服务端发送消息 " + entry.getKey() + " | " + System.currentTimeMillis());
}
Thread.sleep(10000);
}
catch(Exception ex) {
ex.printStackTrace();
}
}
public static void put(Channel channel) {
ALL_CHANNEL.put(channel.id().asShortText(), channel);
}
public static void remove(Channel channel) {
ALL_CHANNEL.remove(channel.id().asShortText());
}
public static void send(Channel channel, Object msg) {
final String textMsg = JSON.toJSONString(msg);
if(channel != null && channel.isActive()) {
TextWebSocketFrame frame = new TextWebSocketFrame(textMsg);
channel.writeAndFlush(frame)
.addListener(new CustomerChannelFutureListener());
}
else {
LOG.error("消息发送失败! textMsg = " + textMsg);
}
}
public static void sendPing(Channel channel) {
if(channel != null && channel.isActive()) {
channel.writeAndFlush(new PingWebSocketFrame())
.addListener(new CustomerChannelFutureListener());
}
else {
LOG.error("消息发送失败! ping");
}
}
public static void sendPong(Channel channel) {
if(channel != null && channel.isActive()) {
channel.writeAndFlush(new PongWebSocketFrame())
.addListener(new CustomerChannelFutureListener());
}
else {
LOG.error("消息发送失败! pong");
}
}
}
channel初始化
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.handler.WebSocketClientFrameHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
import io.netty.handler.timeout.IdleStateHandler;
/**
* ClientChannelInitializer
*
* @date 2019/8/31 16:05
*/
public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
private WebSocketClientHandshaker webSocketClientHandshaker;
private WebSocketClientFrameHandler webSocketFrameHandler;
public ClientChannelInitializer(WebSocketClientHandshaker webSocketClientHandshaker, WebSocketClientFrameHandler
webSocketFrameHandler) {
this.webSocketClientHandshaker = webSocketClientHandshaker;
this.webSocketFrameHandler = webSocketFrameHandler;
}
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new HttpClientCodec());//Http协议编码解码器
pipeline.addLast(new HttpObjectAggregator(65536));//聚合 HttpRequest
pipeline.addLast(new IdleStateHandler(5, 10, 0));
//会处理ping pong close消息
pipeline.addLast(new WebSocketClientProtocolHandler(webSocketClientHandshaker,true));
pipeline.addLast(webSocketFrameHandler);
}
}
客户端主类 : 添加工作组 , 启动websocket client , 建立连接等文章来源:https://www.toymoban.com/news/detail-450767.html
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.handler.WebSocketClientFrameHandler;
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.task.HeartBeatTimerTask;
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.task.ReconnectTimerTask;
import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.thread.NamedThreadFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* SocketClient
*
* @date 2019/8/20 17:53
*/
public class SocketClient {
private static final Logger LOG = LoggerFactory.getLogger(SocketClient.class);
public static String CLIENT_VERSION = "client_version";
private static final int DEFAULT_PORT = 80;
/**
* 长链重连间隔时间,单位s
*/
public static long RECONNECT_INTERVAL = 10;
/**
* 长链心跳时间,单位s
*/
public static long FETCH_PERIOD = 30;
public static String host = "127.0.0.1";
public static int port = 8585;
private Channel channel = null;
private NioEventLoopGroup nioEventLoopGroup;
public AtomicBoolean CHANNEL_IS_READY = new AtomicBoolean(false);
private ScheduledExecutorService RECONNECT_TIMER;
private ScheduledExecutorService HEARTBEAT_TIMER;
static final String URL = System.getProperty("url", "ws://127.0.0.1:8585/boot/imserver/1111");
URI uri = new URI(URL);
static {
// RECONNECT_TIMER = Executors.newSingleThreadScheduledExecutor();
}
public SocketClient(URI uri) throws URISyntaxException {
this.uri = uri;
}
private void start() {
Bootstrap boot = new Bootstrap();
nioEventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
try {
HttpHeaders httpHeaders = new DefaultHttpHeaders();
httpHeaders.add(CLIENT_VERSION, 1);
WebSocketClientHandshaker webSocketClientHandshaker = WebSocketClientHandshakerFactory.newHandshaker(uri,
WebSocketVersion.V13, null, false, httpHeaders);
boot.group(nioEventLoopGroup)
.option(ChannelOption.TCP_NODELAY, true)
.channel(NioSocketChannel.class);
WebSocketClientFrameHandler webSocketFrameHandler = new WebSocketClientFrameHandler();
webSocketFrameHandler.setSocketClient(this);
ClientChannelInitializer clientChannelInitializer =
new ClientChannelInitializer(webSocketClientHandshaker, webSocketFrameHandler);
boot.handler(new LoggingHandler(LogLevel.INFO));
boot.handler(clientChannelInitializer);
port = (uri.getPort() == -1) ? DEFAULT_PORT : uri.getPort();
host = uri.getHost();
channel = boot.connect(host, port).sync().channel();
LOG.info("SocketClient has started. CHANNEL_IS_READY = " + CHANNEL_IS_READY.get());
webSocketFrameHandler.getChannelPromise().sync();
LOG.info("SocketClient has started full. CHANNEL_IS_READY = " + CHANNEL_IS_READY.get());
}
catch(Exception ex) {
ex.printStackTrace();
LOG.error("connect error. uri " + uri.toString());
}
finally {
// nioEventLoopGroup.shutdownGracefully();
}
}
/**
* 客户端连接服务端
*/
public void connect() {
stop();
start();
startReconnect();
doHeartBeat();
}
/**
* 开启线程-断开重连
*/
public void startReconnect() {
RECONNECT_TIMER = new ScheduledThreadPoolExecutor(1,
new NamedThreadFactory("reconnect-schedule-pool", Boolean.TRUE, Thread.NORM_PRIORITY));
// https://www.jianshu.com/p/502f9952c09b
RECONNECT_TIMER.scheduleAtFixedRate(new ReconnectTimerTask(this),
RECONNECT_INTERVAL * 1000L, RECONNECT_INTERVAL * 1000L, TimeUnit.MILLISECONDS);
}
/**
* 心跳
*/
private void doHeartBeat() {
HEARTBEAT_TIMER = new ScheduledThreadPoolExecutor(1,
new NamedThreadFactory("heartbeat-schedule-pool", Boolean.TRUE, Thread.NORM_PRIORITY));
// https://www.jianshu.com/p/502f9952c09b
HEARTBEAT_TIMER.scheduleAtFixedRate(new HeartBeatTimerTask(this),
FETCH_PERIOD * 1000L, FETCH_PERIOD * 1000L, TimeUnit.MILLISECONDS);
}
/**
* 客户端停止
*/
public void stop() {
try {
if(nioEventLoopGroup != null) {
nioEventLoopGroup.shutdownGracefully();
}
if(channel != null) {
channel.close();
}
if(RECONNECT_TIMER != null) {
RECONNECT_TIMER.shutdown();
RECONNECT_TIMER = null;
}
if(HEARTBEAT_TIMER != null) {
HEARTBEAT_TIMER.shutdown();
HEARTBEAT_TIMER = null;
}
}
catch(Exception ex) {
//do nothing.
}
}
public boolean isValid() {
if (channel != null && channel.isActive()) {
return true;
} else {
return false;
}
}
public Channel getChannel() {
return channel;
}
public static void main(String[] args) {
try {
new SocketClient(new URI(URL)).connect();
} catch (URISyntaxException e) {
e.printStackTrace();
}
}
}
重连成功截图:文章来源地址https://www.toymoban.com/news/detail-450767.html
到了这里,关于【netty】java如何作为websocket客户端 对服务端发起请求的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!