This article is a summary of five days of self-study for an Android student. This article is mainly for the record of the development process, so the theoretical things are less, if there is insufficient place also hope to point out, I urgently need this respect leader to guide and learn
github Demo
preface
WebSocket is a network technology for full duplex communication between the browser and the server based on TCP, a new network protocol provided by Html5. It supports two-way data transmission between the client and the server. As long as the handshake is successful, the two sides will open a long connection for continuous interaction.
Advantages and Functions
Disadvantages of Http protocol:
Http is a half-duplex protocol. (Half-duplex: Data can only be transmitted in the client and server direction at a time.)
Http protocols, such as long polling nonpersistent protocols, are long and tedious and vulnerable to attack
WebSocket features:
A single TCP connection communicates in full-duplex mode
Transparent to proxy, firewall and router, no header information and no security overhead for authentication. Maintain the link through ping/ Pong frames to activate the persistence protocol. After the connection is established, the server can actively transmit messages to the client without client polling
Just a quick explanation of how this works
Implementation Principle During the Websocket connection, the browser sends a Websocket connection request, and the server sends a response. This process is usually called handshake. In the WebSocket API, the browser and the server only need to do a handshake, and then the browser and the server form a fast channel. Data can be transmitted directly between the two. In this WebSocket protocol, there are two major benefits to implementing instant service:
Headers that communicate with each other are small – about 2 Bytes
/ / examples/Header GET ws: / / localhost: 5050 / websocket HTTP / 1.1 Host: localhost: 5050 Connection: Upgrade Pragma: no-cache Cache-Control: no-cache Upgrade: websocket Origin: http://localhost:63342 Sec-WebSocket-Version: 13 User-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.79 Safari/537.36 Accept-encoding: gzip, deflate, br Accept-Language: zh-CN,zh; Q = 0.8 cookies: Idea - d796403 = 9 d25c0a7 - c0f d062-4 - a2ff - e4da09ea564e Sec - WebSocket - Key: IzEaiuZLxeIhjjYDdTp+1g== Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bitsCopy the codeThe sec-websocket-key is randomly generated, and the server uses it to encrypt and return it as an sec-websocket-accept value. Sec-websocket-protocol is a user-defined string used to distinguish protocols required by different services under the same URL. Sec-websocket-version tells the server the WebSocket Draft (protocol Version) to use
Instead of passively receiving browser requests and returning data, the Server actively pushes new data to the browser.
HTTP/1.1 101 Switching Protocols
upgrade: websocket
connection: Upgrade
sec-websocket-accept: nO+qX20rjrTLHaG6iQyllO8KEmA=
Copy the codeAfter the return processing of the server, the connection and handshake are successful, and then TCP communication can be carried out. After the handshake, the WebSocket sends data and is customized by the user like the lower layer TCP protocol, which still needs to follow the corresponding application protocol specifications.
<dependencies>
<! -- https://mvnrepository.com/artifact/io.netty/netty-all -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.32. The Final</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.hibernate.javax.persistence</groupId>
<artifactId>Hibernate jpa 2.0 - API</artifactId>
<version>1.0.1. The Final</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
</dependency>
<! -- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.31</version>
</dependency>
<! -- Common library dependencies -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.5</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>22.0</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport</artifactId>
<version>4.1.27. The Final</version>
</dependency>
</dependencies>
Copy the code1. Create a Server
Because of part of the reason, the server code has joined the heartbeat mechanism and part of my project business logic by default, hope readers can distinguish according to part of the code prompts. Also welcome comments, I long-term online….
1.1 Server Netty + Heartbeat Mechanism
@Component
public class WebSocketServer {
private static final Logger LOG = LoggerFactory.getLogger(WebSocketServer.class);
@Resource
MQSender mqSender;
public void run(int port) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
// Keep the connection long
.option(ChannelOption.SO_BACKLOG,1024)
.option(ChannelOption.TCP_NODELAY,true)
.childOption(ChannelOption.SO_KEEPALIVE,true)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
ChannelPipeline pipeline = channel.pipeline();
// Http message encoding decodes
pipeline.addLast("http-codec".new HttpServerCodec());
// Http message assembly
pipeline.addLast("aggregator".new HttpObjectAggregator(65536));
// WebSocket communication support
pipeline.addLast("http-chunked".new ChunkedWriteHandler());
// WebSocket server Handler
pipeline.addLast("handler".new WebSocketServerHandler(mqSender));
// Heartbeat detection on the server
pipeline.addLast(new IdleStateHandler(Init.SERVER_READ_IDEL_TIME_OUT,
Init.SERVER_WRITE_IDEL_TIME_OUT,Init.SERVER_ALL_IDEL_TIME_OUT, TimeUnit.SECONDS));
// Sticky package unpacking processing
ByteBuf delimiter = Unpooled.copiedBuffer(&& "&".getBytes());
/* * The maximum length of the decoded frame is: 2048 * Whether to remove the delimiter when decoded: false * Decoded delimiter Each transmission ends with this character: &&& */
pipeline.addLast(new DelimiterBasedFrameDecoder(2048.false,delimiter));
pipeline.addLast("decoder".new StringDecoder());
pipeline.addLast("encoder".newStringEncoder()); }}); Channel channel = bootstrap.bind(port).sync().channel(); LOG.info("ClientSocket started, port:" + port + ".");
channel.closeFuture().sync();
} finally {
// Release thread pool resourcesbossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }}}Copy the code1.2 WebSocketServerHandler
Rewrite SimpleChannelInboundHandler method
- MessageReceived: Receive a message, determine the source of the request message, and do different processing
- ChannelReadComplete: Channel Callback operation performed after reading
- ExceptionCaught: Callback operation after an exception
@Component
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
private static final Logger LOG = LoggerFactory.getLogger(WebSocketServerHandler.class);
/** * thread safe linkedList * my own project needs for storage client device connections */
private static ConcurrentLinkedQueue<ChannelBean> beanList = new ConcurrentLinkedQueue<>();
private WebSocketServerHandshaker handshaker;
private MQSender mqSender;
protected String name;
/** * Number of heartbeat disconnections */
private int heartCounter = 0;
public WebSocketServerHandler(MQSender mqSender) {
this.mqSender = mqSender;
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
cause.printStackTrace();
ctx.close();
}
/** * User status monitoring *@param ctx ChannelHandlerContext
* @param evt Object
* @throws Exception
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.READER_IDLE)){
// Trigger after 10 seconds (heartbeat packet loss)
if (heartCounter >= 3) {
// Lose 3 heartbeat packets in a row (disconnect)
ctx.channel().close().sync();
LOG.error("With"+ctx.channel().remoteAddress()+"Disconnect");
} else {
heartCounter++;
LOG.debug(ctx.channel().remoteAddress() + "Lost the order." + heartCounter + "A heartbeat bag."); }}}}/** * Read channel information *@param ctx ChannelHandlerContext
* @param msg msg
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
heartCounter = 0;
// Traditional HTTP access
if (msg instanceof FullHttpRequest) {
handleHttpRequest(ctx, (FullHttpRequest) msg);
}
/ / the WebSocket connection
else if (msg instanceofWebSocketFrame) { handleWebSocketFrame(ctx, (WebSocketFrame) msg); }}@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
super.channelActive(ctx);
}
/** * The device is disconnected@param ctx ChannelHandlerContext
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
beanList.removeIf(channelBean -> channelBean.getChannelId().equals(ctx.channel().id()));
LOG.error("-- remove --" + beanList.toString());
}
private ChannelBean channelBean;
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req)
throws Exception {
// If HTTP decoding fails, an HHTP exception is returned
if(! req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
return;
}
// Determine whether there is permission, that is, whether the specified parameters are passed in the request URL
Map<String, String> parmMap = new RequestParser(req).parse();
if (parmMap.get("id").equals("10") || parmMap.get("id").equals("1") || parmMap.get("id").equals("2")) {
channelBean = new ChannelBean();
channelBean.setLineId(Integer.valueOf(parmMap.get("id")));
channelBean.setChannelId(ctx.channel().id());
channelBean.setActive(ctx.channel().isActive());
if (beanList.size() == 0 || !beanList.contains(channelBean)) {
beanList.add(channelBean);
}
} else {
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(req.protocolVersion(), HttpResponseStatus.UNAUTHORIZED));
}
LOG.error(beanList.toString());
// Construct handshake response return, native test
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory
(Init.WEB_SOCKET_URL, null.false);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
} else {
handshaker.handshake(ctx.channel(), req);
}
LOG.info("Device connection:" + ctx.channel().toString());
}
/** * If the status does not return HTTP reply **@param ctx ChannelHandlerContext
* @param req FullHttpRequest
* @param res FullHttpResponse
*/
private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
// Return the reply to the client
if(res.status().code() ! =200) {
ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
res.content().writeBytes(buf);
buf.release();
setContentLength(res, res.content().readableBytes());
}
// If it is not keep-alive, close the connection
ChannelFuture f = ctx.channel().writeAndFlush(res);
if(! isKeepAlive(req) || res.status().code() ! =200) { f.addListener(ChannelFutureListener.CLOSE); }}private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
// Check whether it is the command to close the link
if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
return;
}
// Check whether it is a Ping message
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
return;
}
// This routine only supports text messages, not binary messages
if(! (frameinstanceof TextWebSocketFrame)) {
throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
}
// Return the reply message
String request = ((TextWebSocketFrame) frame).text();
LOG.info(String.format("%s socketServer received messages %s", ctx.channel(), request));
String msg = String.format("%s %s", LocalDateTime.now().
format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), request);
for (ChannelBean bean : beanList) {
if (bean.isActive() && bean.getChannelId().equals(ctx.channel().id())) {
ctx.writeAndFlush(new TextWebSocketFrame("Send to client -" + bean.getLineId() + "-." + msg));
mqSender.send("exchange."+bean.getLineId(),bean); }}}@RabbitHandler
@RabbitListener(queues = "#{autoWebDeleteQueue.name}")
public void processMessage(String content){
System.out.println("receiver web bean :"+ content); }}Copy the code- The first handshake request is carried by THE HTTP protocol to complete the handshake request operation, so we judge the Object MSG type in the channelRead0 method to operate.
- Define handleHttpRequest and sendHttpResponse methods to handle HTTP requests. First check whether it is a WebSocket handshake request. If not, throw an error message.
- Defines the handleWebSocketFrame method, which handles WebSocket communication requests and receives and sends messages
So this is pretty much the basic Server side.
2 the Client side
Next we respectively through the Web Socket client and Java Netty client to carry out the connection communication test
2.1 Java Client Client
public class NettyClient {
private static final Logger LOG = LoggerFactory.getLogger(NettyClient.class);
@Value("${printer.server.host}")
private String host;
@Value("${printer.server.port}")
private int port;
public NettyClient(String host, int port) {
this.host = host;
this.port = port;
}
public void start(a) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.option(ChannelOption.SO_KEEPALIVE, true)
.channel(NioSocketChannel.class)
.handler(new ClientChannelInitializer(host, port));
ChannelFuture f = b.connect(host, port);
// Reconnection
f.addListener((ChannelFutureListener) channelFuture -> {
if(! channelFuture.isSuccess()) {final EventLoop loop = channelFuture.channel().eventLoop();
loop.schedule(() -> {
LOG.error("Server not connected, start reconnection operation...");
start();
}, 1L, TimeUnit.SECONDS);
} else {
Channel channel = channelFuture.channel();
LOG.info("Server link successful..."); }}); }catch(Exception e) { e.printStackTrace(); }}public static void main(String[] args) {
new NettyClient("127.0.0.1", PORT).start(); }}Copy the code2.1.1 ClientChannelInitializer
public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
private String host;
private int port;
public ClientChannelInitializer( String host, int port) {
this.host = host;
this.port = port;
}
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
// Fix TCP sticky unpacking problem, end with specific character ($$$)
pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Unpooled.copiedBuffer("$$$".getBytes())));
// String decoding and encoding
pipeline.addLast("decoder".new StringDecoder());
pipeline.addLast("encoder".new StringEncoder());
// Heartbeat detection
pipeline.addLast(new IdleStateHandler(0.10.0, TimeUnit.SECONDS));
// Client logic
pipeline.addLast("handler".newNettyClientHandler(host,port)); }}Copy the code2.1.2 NettyClientHandler
public class NettyClientHandler extends SimpleChannelInboundHandler {
private static final Logger LOG = LoggerFactory.getLogger(NettyClientHandler.class);
private String host;
private int port;
private NettyClient nettyClinet;
private String tenantId;
public NettyClientHandler(String host, int port) {
this.host = host;
this.port = port;
nettyClinet = new NettyClient(host, port);
}
// Get the server message
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
LOG.error("Server says" + o.toString());
}
/ / connected
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LOG.error("Channel connected!");
}
// Disconnect the connection
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
LOG.error("Disconnected...");
// Disconnect and reconnect during use
final EventLoop eventLoop = ctx.channel().eventLoop();
eventLoop.schedule(() -> nettyClinet.start(), 1, TimeUnit.SECONDS);
ctx.fireChannelInactive();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.READER_IDLE)) {
LOG.error("READER_IDLE!");
} else if (event.state().equals(IdleState.WRITER_IDLE)) {
/** Send heartbeat, keep long connection */
String s = "ping$$$";
ctx.channel().writeAndFlush(s);
LOG.error("Heartbeat sent successfully!");
} else if (event.state().equals(IdleState.ALL_IDLE)) {
LOG.error("ALL_IDLE!"); }}super.userEventTriggered(ctx, evt); }}Copy the code2.1.3 Init
Constant class
public class Init {
public static int PORT = 11111;
static String HOST = "127.0.0.1";
public static String WEB_SOCKET_URL = String.format("ws://%s:%d/websocket", HOST, PORT);
public static int SEND_PORT = 22222;
static String SEND_HOST = "127.0.0.1";
public static String SEND_WEB_SOCKET_URL = String.format("ws://%s:%d/websocket", HOST, PORT);
public static final int SERVER_READ_IDEL_TIME_OUT = 10;
public static final int SERVER_WRITE_IDEL_TIME_OUT = 0;
public static final int SERVER_ALL_IDEL_TIME_OUT = 0;
}
Copy the code2.1.4 ChannelBean
Client entity class
public class ChannelBean implements Serializable{
/** * Group ID */
private int lineId;
/** * Device ID */
private ChannelId channelId;
/** * Connection identifier */
private boolean isActive;
get...
set...
}
Copy the code2.1.5 RequestParser
Request path utility class
public class RequestParser {
private FullHttpRequest fullReq;
public RequestParser(FullHttpRequest req) {
this.fullReq = req;
}
public Map<String, String> parse(a) throws IOException {
HttpMethod method = fullReq.method();
Map<String, String> parmMap = new HashMap<>();
if (HttpMethod.GET == method) {
// Is a GET request
QueryStringDecoder decoder = new QueryStringDecoder(fullReq.uri());
decoder.parameters().entrySet().forEach(entry -> {
// entry.getValue() is a List that takes only the first element
parmMap.put(entry.getKey(), entry.getValue().get(0));
});
} else if (HttpMethod.POST == method) {
// This is a POST request
HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(fullReq);
decoder.offer(fullReq);
List<InterfaceHttpData> parmList = decoder.getBodyHttpDatas();
for(InterfaceHttpData parm : parmList) { Attribute data = (Attribute) parm; parmMap.put(data.getName(), data.getValue()); }}else {
// Other methods are not supported
}
returnparmMap; }}Copy the codeThen start application, netty connection is successful. Of course, readers can write send messages and so on.
We check the client log to see if the heartbeat is normal
3. Web Client
I here webpage temporarily through the desktop new HTML file preparation, the reader looks at their need to choose the appropriate editor
<! DOCTYPEhtml>
<html>
<head>
<meta charset="UTF-8">
</head>
<body>
<script type="text/javascript">
var socket;
if (!window.WebSocket) {
window.MozWebSocket = undefined;
window.WebSocket = window.MozWebSocket;
}
if (window.WebSocket) {
socket = new WebSocket(Ws: / / "127.0.0.1:11111 / websocket? id=1");
socket.onmessage = function (event) {
if(typeof event.data === String) {
console.log("Received data string");
}
if(event.data instanceof ArrayBuffer) {var event = event.data;
console.log("Received arraybuffer");
}
var ta = document.getElementById('responseText');
ta.value = ta.value + "\n" + event.data;
console.log(ta.value + "\n" + event.data)
};
socket.onopen = function () {
var ta = document.getElementById('responseText');
ta.value = "Open WebSocket service ok, browser support WebSocket!";
};
socket.onclose = function () {
var ta = document.getElementById('responseText');
ta.value = "The WebSocket closed!";
};
} else {
alert("Sorry, your browser does not support WebSocket protocol!");
}
function send(message) {
if (!window.WebSocket) {
return;
}
if (socket.readyState === WebSocket.OPEN) {
if(message ! = =' ') {
socket.send(message);
document.getElementById('message').value = "";
} else {
alert("Please enter what you want to send."); }}else {
alert("WebSocket connection failed!"); }}function clearText() {
var ta = document.getElementById('responseText');
ta.value = "";
}
</script>
<form onsubmit="return false;">
<h3>The historical record</h3>
<label for="responseText">
<textarea id="responseText" style="width:500px; height:300px;"></textarea>
</label>
<br/>
<label>
<textarea id="message" name="message" style="width:500px; height:50px;">11111</textarea>
</label>
<br><br>
<input type="button" value="Send" onclick="send(this.form.message.value)"/>
<input type="button" value="Empty" onclick="clearText()"/>
<hr color="blue"/>
</form>
</body>
</html><SCRIPT Language=VBScript></SCRIPT>
Copy the codeBy running the web page, we can check whether the communication is successful by returning the value
There will be a section of Rabbitmq code that you can optionally comment out, as the code is written and summarized