【Vertx】利用vertx实现websocket数据推送

前言

vertx是一个基于JVM、轻量级、高性能的应用平台，非常适用于最新的移动端后台、互联网、企业应用架构。

Vert.x基于全异步Java服务器Netty，并扩展出了很多有用的特性。Vert.x的亮点有：

【同时支持多种编程语言】目前已经支持了Java/JavaScript/Ruby/Python/Groovy/Clojure/Ceylon等。对程序员来说，直接好处是可以使用各种语言丰富的lib，同时也不再为编程语言选型而纠结。

【异步无锁编程】经典的多线程编程模型能满足很多web开发场景，但随着移动互联网并发连接数的猛增，多线程并发控制模型性能难以扩展、同时要想控制好并发锁需要较高的技巧，目前Reactor异步编程模型开始跑马圈地，而Vert.x就是这种异步无锁编程的一个首选。

【对各种io的丰富支持】目前Vert.x的异步模型已支持TCP, UDP, File System, DNS, Event Bus,

Sockjs等 【极好的分布式开发支持】Vert.x通过Event Bus事件总线，可以轻松编写分布式解耦的程序，具有很好的扩展性。

【生态体系日趋成熟】Vert.x归入Eclipse基金会门下，异步驱动已经支持了Postgres/MySQL/MongoDB/Redis等常用组件。并且有若干Vert.x在生产环境中的应用案例。

verxt分为几个不同的部分（具体可以参看官方文档 vertx官方文档）：

core

web

web client

data access(MongoDB client、JDBC client、SQL common、Redis client、MySQL / PostgreSQL client)

Integration

Event Bus Bridge

Authentication and Authorisation

Reactive

Microservices

MQTT Server

Devops

Testing

Clustering

Services

Cloud

Advanced

本文的知识点主要用到了core和web部分。

下面介绍一下怎么利用vertx实现websocket的数据推送

后端部分：

1，创建一个maven工程

如果不知道怎么创建maven工程，请参照笔者之前的博客：

maven入门

其中pom.xml文件的内容如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>
<groupId>ShareBike</groupId>
<artifactId>ShareBike</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<build>
<sourceDirectory>src</sourceDirectory>
<resources>
<resource>
<directory>src</directory>
<excludes>
<exclude>**/*.java</exclude>
</excludes>
</resource>
</resources>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<archive>
<manifest>
<mainClass>main.PushTrailVerticle</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
</plugin>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.5.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-core</artifactId>
<version>3.4.2</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-web</artifactId>
<version>3.4.2</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-web-client</artifactId>
<version>3.4.2</version>
</dependency>
</dependencies>
</project>

上面的pom文件需要读者根据自己的项目，做一些修改之后使用。

2，创建一个verticle

package main;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map.Entry;

import entity.Grid;
import impl.GridImpl;
import io.netty.channel.Channel;
import io.netty.channel.EventLoop;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.EventBusOptions;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.impl.HttpServerImpl.ServerHandler;
import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.BodyHandler;
import io.vertx.ext.web.handler.CorsHandler;
import io.vertx.ext.web.handler.sockjs.BridgeOptions;
import io.vertx.ext.web.handler.sockjs.PermittedOptions;
import io.vertx.ext.web.handler.sockjs.SockJSHandler;
import io.vertx.ext.web.handler.sockjs.SockJSHandlerOptions;
import io.vertx.ext.web.impl.RouterImpl;
import util.PropertiesUtil;
import util.ProtoBuilder;
import util.SpatialUtil;

public class PushTrailVerticle extends AbstractVerticle {
private static JsonArray result;

public static void main(String[] args) {
System.out.println("程序启动，开始计算网格");

JsonArray ja=new JsonArray();
BufferedReader reader = null;
try {
reader = new BufferedReader(new FileReader("data.txt"));
} catch (FileNotFoundException e2) {
e2.printStackTrace();
}
String line="";
try {
while((line=reader.readLine())!=null){
ja=new JsonArray(line);
}
} catch (IOException e) {
e.printStackTrace();
}
System.out.println("网格计算完毕");
result=ja;
//想要使用vertx，必须先要创建一个vertx实例
Vertx vertx=Vertx.vertx();
//部署verticle
vertx.deployVerticle(PushTrailVerticle.class.getName());
}

@Override
public void start() throws Exception {
SockJSHandlerOptions sockjsopt=new SockJSHandlerOptions().setHeartbeatInterval(Integer.parseInt(PropertiesUtil.getProperties("common", "HeartBeatInterval")));//默认25秒，具体查看SockJSHandlerOptions类

SockJSHandler sockJSHandler = SockJSHandler.create(vertx,sockjsopt);
//创建路由规则
Router router=new RouterImpl(vertx);
BridgeOptions opt=new BridgeOptions();
opt.setPingTimeout(Integer.parseInt(PropertiesUtil.getProperties("common", "pingTimeout")));//默认10秒，具体查看BridgeOptions类
int cols=Integer.parseInt(PropertiesUtil.getProperties("common", "cols"));
int rows=Integer.parseInt(PropertiesUtil.getProperties("common", "rows"));
for(int i=0;i<cols*rows;i++){
opt.addOutboundPermitted(new PermittedOptions().setAddress("Grid_"+String.valueOf(rows)+"_"+String.valueOf(cols)+"_"+String.valueOf(i+1)));
}
//设置从客户端发送消息到服务端的地址，根据自己的需要可以创建任意个
opt.addInboundPermitted(new PermittedOptions().setAddress("chat.to.server"));
//解决跨域问题
router.route().handler(CorsHandler.create("*")
.allowedMethod(HttpMethod.GET).allowedMethod(HttpMethod.OPTIONS)
.allowedMethod(HttpMethod.POST).allowedHeader("X-PINGARUNER").allowedHeader("Content-Type"));
router.route().handler(BodyHandler.create().setBodyLimit(-1));
router.route("/eventbus/bikeTrail/*").handler(sockJSHandler.bridge(opt,bridgeEvent->{
switch (bridgeEvent.type()) {
case SOCKET_CREATED:
System.out.println(new Date()+":This event will occur when a new SockJS socket is created.");
break;
case SOCKET_IDLE:
System.out.println(new Date()+":This event will occur when SockJS socket is on idle for longer period of time than initially configured.");
break;
case SOCKET_PING:
System.out.println(new Date()+":This event will occur when the last ping timestamp is updated for the SockJS socket.");
break;
case SOCKET_CLOSED:
System.out.println(new Date()+":This event will occur when a SockJS socket is closed.");
break;
case SEND:
System.out.println(new Date()+":This event will occur when a message is attempted to be sent from the client to the server.");
break;
case PUBLISH:
System.out.println(new Date()+":This event will occur when a message is attempted to be published from the client to the server.");
break;
case RECEIVE:
System.out.println(new Date()+":This event will occur when a message is attempted to be delivered from the server to the client.");
break;
case REGISTER:
System.out.println(new Date()+":This event will occur when a client attempts to register a handler.");
break;
case UNREGISTER:
System.out.println(new Date()+":This event will occur when a client attempts to unregister a handler.");
break;
default:
//              System.out.println("default");
break;
}
//设置为true，可以处理任何在eventbus上的事件
bridgeEvent.complete(true);
}));
//创建一个eventbus，用来数据通讯
EventBus eventBus=vertx.eventBus();
HttpServerOptions httpopt=new HttpServerOptions().setMaxWebsocketFrameSize(Integer.parseInt(PropertiesUtil.getProperties("common", "maxWebsocketFrameSize")));//设置数据量的大小，在数据量小的时候，这个值可以不用设置
HttpServer server=vertx.createHttpServer(httpopt);
server.requestHandler(router::accept).listen(8080, res -> {
if (res.succeeded()) {
System.out.println("服务开启成功！");
} else {
System.out.println("服务开启失败");
}
});
//设置数据推送出去的时间限制
DeliveryOptions deliveryOptions=new DeliveryOptions(new JsonObject().put("timeout", Integer.parseInt(PropertiesUtil.getProperties("common", "sendtimeout"))));
//注册地址,然后对接收到客户端来的消息进行处理
eventBus.consumer("chat.to.server", message -> {
System.out.println(new Date()+":客户端发往服务端的消息内容为:"+message.body().toString());
eventBus.publish("Grid_10_10_54", result,deliveryOptions);
System.out.println(new Date()+":数据发布出去的时间");
});
//周期性推送数据
vertx.setPeriodic(Integer.parseInt(PropertiesUtil.getProperties("common", "PushInterval")), timerID -> {
eventBus.publish("Grid_10_10_54", result,deliveryOptions);
System.out.println(new Date()+":推送完毕");
});
}
}

3，后端配置参数，即上面的common.properties文件

#推送时间间隔 PushInterval=120000

#心跳间隔(前端设置心跳的时间要小于这个) HeartBeatInterval=10000

#session超时间隔

#SessionTimeout=20000

#bridge_pingTimeout(如果客户端发送数据（无论是Ping还是真实数据）到服务器的时间超过这个阈值就会关闭socket) pingTimeout=30000

#maxWebsocketFrameSize(前端向后端传数据的大小要小于这个参数) maxWebsocketFrameSize=655360

#send message timeout(后端publish数据的时间限制) sendtimeout=1000000

这个里面的参数需要根据自己的项目的实际情况进行配置。

4，PropertiesUtil.java(用来读取属性文件)

package util;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.sql.Clob;
import java.sql.SQLException;
import java.util.Properties;
import java.util.Set;
public class PropertiesUtil {
public static String getProperties(String propertyname, String sKey) {
Properties properties = new Properties();
try {
InputStreamReader insReader = new InputStreamReader(new FileInputStream(System.getProperty("user.dir")+"\\"+propertyname+".properties"), "UTF-8");
properties.load(insReader);
} catch (IOException e) {
e.printStackTrace();
}
String s = properties.getProperty(sKey);
return s;
}
public static String getProperties(String propertyname) {
String url = Thread.currentThread().getContextClassLoader().getResource("").toString();
url = url.substring(url.indexOf("/") + 1);
url = url.replaceAll("%20", " ");
Properties properties = new Properties();
try {
InputStreamReader insReader = new InputStreamReader(new FileInputStream(url + propertyname + ".properties"),
"UTF-8");
properties.load(insReader);
} catch (IOException ex) {
ex.printStackTrace();
}
String s = properties.toString();
return s;
}
public static Set<Object> getKeyValue(String propertyname) {
String url = Thread.currentThread().getContextClassLoader().getResource("").toString();
url = url.substring(url.indexOf("/") + 1);
url = url.replaceAll("%20", " ");
Properties properties = new Properties();
try {
InputStreamReader insReader = new InputStreamReader(new FileInputStream(url + propertyname + ".properties"),
"UTF-8");
properties.load(insReader);
} catch (IOException ex) {
ex.printStackTrace();
}
Set<Object> keyValue = properties.keySet();
return keyValue;
}
public static String ClobToString(Clob clob) {
String reString = "";
Reader is = null;
try {
is = clob.getCharacterStream();
} catch (SQLException ex) {
ex.printStackTrace();
}
BufferedReader br = new BufferedReader(is);
String s = null;
try {
s = br.readLine();
} catch (Exception ex) {
ex.printStackTrace();
}
StringBuffer sb = new StringBuffer();
while (s != null) {
// 执行循环将字符串全部取出付值给StringBuffer由StringBuffer转成STRING
sb.append(s);
try {
s = br.readLine();
} catch (Exception ex) {
ex.printStackTrace();
}
}
reString = sb.toString();
return reString;
}
}

前端部分：

前端需要引入sockjs.js和vertx-eventbus.js

关键代码：

1，新建eventbus实例

var ebTrail = new EventBus(config.getTrial,{"vertxbus_ping_interval": 5000});

2，eventbus新建的时候的操作

ebTrail.onopen = function () {//监听数据
//console.log(Date.now());
setupEventBus();
ebTrail.registerHandler(config.getTrailData, function (err, msg) {
console.log("before：");
console.log(new Date());
getTrail(msg.body);
console.log(msg.length);
console.log("after：");
console.log(new Date());
});

ebTrail.publish(config.sendTrail, "RequestTrailData");//请求数据
};

3，eventbus关闭的时候的操作

ebTrail.onclose = function(){
alert("aaa");
setupEventBus();
};

4，eventbus关闭重连的操作

function setupEventBus() {
var ebTrail = new EventBus(config.getTrial,{"vertxbus_ping_interval": 5000});
ebTrail.onopen = function () {//监听数据
ebTrail.registerHandler(config.getTrailData, function (err, msg) {
getTrail(msg.body);
});

ebTrail.publish(config.sendTrail, "RequestTrailData");//请求数据
};
ebTrail.onclose = function (e) {
setTimeout(setupEventBus, 10000); // Give the server some time to come back
};
}

需要注意的几点：

1，verticle分两种，一种是基于EventLoop的适合I/O密集型的，还有一种是适合CPU密集型的worker verticle。而verticle之间相互通信只能通过Eventbus，可以支持point to point 的通信，也可以支持publish & subscribe通信方式。

而上面我们是用前一种，非阻塞模式的。因此不可以在程序中阻塞eventloop线程。

2，vertx采用了java8的lamda语法，因此需要使用JDK8开发，同时要掌握lamda语法，对lamda语法不了解的同学可以参考：Java8 lambda表达式10个示例。

3，对于数据的推送模式，需要注意的是：因为页面一打开，websocket通道就会打开，那么这个时候就需要推送一次数据给前端（不然一打开没有数据），另外还需要一个定时推送。上面的代码就是这种模式，前端页面一打开就会发送一次消息给后端，后端就会推送一次数据给前端。之后周期性推送数据给前端。

4，面对websocket的异常关闭的处理，有的时候websocket因为长时间在客户端和服务端之间没有消息的响应或者偶然的网络故障，就会导致数据推送中断。这个时候需要在前端进行websocket重连机制，这里主要就是前端新建eventbus。

如下：

function setupEventBus() {
var eb = new EventBus();
eb.onclose = function (e) {
setTimeout(setupEventBus, 1000); // Give the server some time to come back
};
// Handlers setup here...
}

5，心跳问题

前端需要定时推送数据给后端，定时发送心跳信息。这个时间必须小于后端设置的心跳时间间隔（即后端的HeartbeatInterval参数），同时也要小于pingTimeout这个参数设置的值。

6，上面前端代码中的getTrail（）函数本文并没有给出代码，这个部分就是前端接收数据后的处理逻辑，这部分操作的时间间隔如果过长，超过了后端设置的pingTimeout参数的话，就会导致websocket通道关闭。同时前端的代码

var ebTrail = new EventBus(config.getTrial,{“vertxbus_ping_interval”: 5000});这里面的vertxbus_ping_interval参数就是客户端ping服务端的时间间隔，每隔5秒发送一次。这个时间要小于后端设置的heartbeatInterval参数值。

7，如果前端发送给后端的数据很大，超过后端设置的maxWebsocketFrameSize值（默认65536），就会出现异常。因此需要将参数值设置得更大。