I have seven solutions for implementing real-time web messaging

I have a friend~

I have built a small website, and now I want to implement a function of pushing web messages within the website. Yes, it is the little red dot in the picture below, which is a very commonly used function.

But he hasn't figured out how to do it yet, so I helped him organize several solutions and implemented them simply.

What is push?

There are many scenarios for push notifications. For example, if someone follows my official account, I will receive a push message to attract me to click and open the app.

Message push usually refers to the active message push to the user's current web page or mobile device APP by website operations staff through some tools.

Message push is generally divided into web-side message push and mobile-side message push.

The above type belongs to mobile message push. Common web message push includes in-site messages, number of unread emails, number of monitoring alarms, etc., and is also widely used.

Before we get into the specific implementation, let's analyze the previous requirements again. In fact, the function is very simple. As long as an event is triggered (resources are actively shared or messages are actively pushed in the background), the notification red dot on the web page will be +1 in real time.

Usually there are several message push tables on the server side, which are used to record different types of messages pushed by users when they trigger different events. The front end actively queries (pull) or passively receives (push) the number of all unread messages of the user.

There are two forms of message push: push and pull. Let's take a look at them one by one.

Short Polling

Polling is probably the simplest way to implement message push. Here we will temporarily divide polling into short polling and long polling.

Short polling is easy to understand. At a specified time interval, the browser sends an HTTP request to the server. The server returns the unread message data to the client in real time, and the browser then renders and displays it.

A simple JS timer can do the job. Request the unread message count interface once a second and display the returned data.

 setInterval ( ( ) => {
 // Method request
 messageCount ( ) .then ( ( res ) => {
 if ( res .code === 200 ) {
 this .messageCount = res .data
 }
 } )
 } , 1000 ) ;

The effect is still acceptable. Although short polling is simple to implement, its shortcomings are also obvious. Since the pushed data does not change frequently, the client will make a request regardless of whether there is a new message generated by the backend at this time, which will inevitably cause great pressure on the server and waste bandwidth and server resources.

Long Polling

Long polling is an improved version of the above short polling. It can reduce the waste of server resources as much as possible while ensuring the relative real-time nature of messages. Long polling is widely used in middleware, such as Nacos and apollo configuration centers, and message queues such as kafka and RocketMQ.

Is the interaction model of Nacos configuration center push or pull? In the article, I introduced the implementation principle of Nacos long polling in detail. Interested friends can take a look.

This time I used the apollo configuration center to implement long polling and applied a class called DeferredResult, which is an asynchronous request mechanism provided by Spring after servelet 3.0, which literally means delayed results.

DeferredResult​ allows the container thread to quickly release occupied resources without blocking the request thread, thereby accepting more requests and improving the system throughput. It then starts the asynchronous worker thread to process the actual business logic. After processing is complete, it calls DeferredResult.setResult(200) to submit the response result.

Next we use long polling to implement message push.

Because one ID may be monitored by multiple long polling requests, I use the Multimap structure provided by the guava package to store long polling. One key can correspond to multiple values. Once a key change is detected, all corresponding long polls will respond. The front end gets the status code of non-request timeout, knows the data change, actively queries the unread message count interface, and updates the page data.

 @Controller
 @RequestMapping ( "/polling" )
 public class PollingController {

 // Store the long polling collection that monitors a certain ID
 // Thread synchronization structure
 public static Multimap < String , DeferredResult < String >> watchRequests = Multimaps .synchronizedMultimap ( HashMultimap .create ( ) ) ;

 /**
 * Public account: Programmer Xiaofu
 * Set up monitoring
 */
 @GetMapping ( path = "watch/{id}" )
 @ResponseBody
 public DeferredResult < String > watch ( @PathVariable String id ) {
 // Delay object setting timeout
 DeferredResult < String > deferredResult = new DeferredResult <> ( TIME_OUT ) ;
 // Remove the key when the asynchronous request is completed to prevent memory overflow
 deferredResult .onCompletion ( ( ) -> {
 watchRequests .remove ( id , deferredResult ) ;
 } ) ;
 // Register long polling request
 watchRequests .put ( id , deferredResult ) ;
 return deferredResult ;
 }

 /**
 * Public account: Programmer Xiaofu
 * Change data
 */
 @GetMapping ( path = "publish/{id}" )
 @ResponseBody
 public String publish ( @PathVariable String id ) {
 // Data changes take out all long polling requests of the monitoring ID and respond to them one by one
 if ( watchRequests .containsKey ( id ) ) {
 Collection < DeferredResult < String >> deferredResults = watchRequests .get ( id ) ;
 for ( DeferredResult < String > deferredResult : deferredResults ) {
 deferredResult .setResult ( "I updated" + new Date ( ) ) ;
 }
 }
 return "success" ;
 }

When the request exceeds the set timeout period, an AsyncRequestTimeoutException​ exception will be thrown. Here, you can directly use @ControllerAdvice to capture and return it globally. After the front end obtains the agreed status code, it will initiate a long polling request again, and the call will be repeated.

 @ControllerAdvice
 public class AsyncRequestTimeoutHandler {

 @ResponseStatus ( HttpStatus .NOT_MODIFIED )
 @ResponseBody
 @ExceptionHandler ( AsyncRequestTimeoutException.class )
 public String asyncRequestTimeoutHandler ( AsyncRequestTimeoutException e ) {
 System .out .println ( "Asynchronous request timed out" ) ;
 return "304" ;
 }
 }

Let's test it. First, the page initiates a long polling request /polling/watch/10086 to monitor message changes. The request is suspended and the data is not changed until it times out. A long polling request is initiated again. Then the data is manually changed to /polling/publish/10086. The long polling gets a response. After the front-end completes the business logic, the request is initiated again, and the cycle repeats.

Long polling has much better performance than short polling, but it still generates more requests, which is one of its imperfections.

iframe stream

The iframe stream is to insert a hidden <iframe> tag into the page, and create a long connection between the server and the client by requesting the message quantity API interface in src, so that the server continues to transmit data to the iframe.

“

The data transmitted is usually HTML​ or embedded javascript scripts to achieve the effect of real-time page updates.

​

This method is simple to implement, and only one <iframe> tag is needed on the front end.

 < iframe src = "/iframe/message" style = "display:none" ></ iframe >

The server can directly assemble HTML and JS script data and write it to the response.

 @Controller
 @RequestMapping ( "/iframe" )
 public class IframeController {
 @GetMapping ( path = "message" )
 public void message ( HttpServletResponse response ) throws IOException , InterruptedException {
 while ( true ) {
 response .setHeader ( "Pragma" , "no-cache" ) ;
 response .setDateHeader ( "Expires" , 0 ) ;
 response .setHeader ( "Cache-Control" , "no-cache,no-store" ) ​​;
 response .setStatus ( HttpServletResponse .SC_OK ) ;
 response .getWriter ( ) .print ( " <script type=\"text/javascript\">\n" +
 "parent.document.getElementById('clock').innerHTML = \"" + count .get ( ) + "\";" +
 "parent.document.getElementById('count').innerHTML = \"" + count .get ( ) + "\";" +
 "</script>" ) ;
 }
 }
 }

But I personally don’t recommend it, because it will show on the browser that the request has not been loaded, and the icon will keep spinning, which is simply a killer for obsessive-compulsive disorder.

SSE (My Way)

Many people may not know that in addition to the familiar mechanism of WebSocket​, the server can also use server-sent events (SSE) to push messages to the client.

SSE is based on the HTTP protocol. We know that the HTTP protocol in the general sense cannot enable the server to actively push messages to the client, but SSE is an exception. It changes the idea.

SSE opens a one-way channel between the server and the client. The server no longer responds with a one-time data packet but with a text/event-stream type of data stream information, which is streamed from the server to the client when there is a data change.

The overall implementation idea is somewhat similar to online video playback. The video stream will be pushed to the browser continuously. You can also understand it as the client completing a download that takes a long time (poor network connection).

SSE​ and WebSocket have similar functions. They can both establish communication between the server and the browser and enable the server to push messages to the client. However, there are some differences:

• SSE is based on the HTTP protocol, and they do not require special protocols or server implementations to work; WebSocket requires a separate server to handle the protocol.
• SSE is a one-way communication, which can only be done from the server to the client; webSocket is a full-duplex communication, which means that both parties can send and receive information at the same time.
• SSE is simple to implement and has low development costs, and does not require the introduction of other components; WebSocket transmission data requires secondary parsing, and the development threshold is higher.
• SSE supports disconnection reconnection by default; WebSocket needs to be implemented by yourself.
• SSE can only transmit text messages, and binary data needs to be encoded before transmission; WebSocket supports the transmission of binary data by default.

How to choose between SSE and WebSocket?

There is no good or bad technology, only which one is more suitable

SSE seems to have been less well known, partly because of the emergence of WebSockets, which provides a richer protocol for performing two-way, full-duplex communication. For games, instant messaging, and scenarios that require two-way near-real-time updates, having a two-way channel is more attractive.

However, in some cases, data does not need to be sent from the client. Instead, you only need some updates of server operations. For example, in-site messages, number of unread messages, status updates, stock quotes, monitoring quantity and other scenarios, SEE​ has more advantages in terms of ease of implementation and cost. In addition, SSE​ has many features that WebSockets​ lacks in design, such as: automatic reconnection​, event ID​, and the ability to send arbitrary events.

The front end only needs to make an HTTP request with a unique ID, open the event stream, and listen to the events pushed by the server.

 < script >
 let source = null ;
 let userId = 7777
 if ( window .EventSource ) {
 // Establish a connection
 source = new EventSource ( 'http://localhost:7777/sse/sub/' + userId ) ;
 setMessageInnerHTML ( "connecting user=" + userId ) ;
 /**
 * Once the connection is established, the open event will be triggered
 * Another way to write: source.onopen = function (event) {}
 */
 source .addEventListener ( 'open' , function ( e ) {
 setMessageInnerHTML ( "Establishing connection..." ) ;
 } , false ) ;
 /**
 * The client receives data from the server
 * Another way to write: source.onmessage = function (event) {}
 */
 source .addEventListener ( 'message' , function ( e ) {
 setMessageInnerHTML ( e.data ) ;
 } ) ;
 } else {
 setMessageInnerHTML ( "Your browser does not support SSE" ) ;
 }
 </ script >

The server-side implementation is simpler. Create an SseEmitter object and put it into sseEmitterMap for management.

 private static Map < String , SseEmitter > sseEmitterMap = new ConcurrentHashMap <> ( ) ;

 /**
 * Create a connection
 *
 * @date: 2022/7/12 14:51
 * @auther: Public account: Programmer Xiaofu
 */
 public static SseEmitter connect ( String userId ) {
 try {
 // Set the timeout period. 0 means no expiration. Default is 30 seconds
 SseEmitter sseEmitter = new SseEmitter ( 0 L ) ;
 // Register callback
 sseEmitter .onCompletion ( completionCallBack ( userId ) ) ;
 sseEmitter .onError ( errorCallBack ( userId ) ) ;
 sseEmitter .onTimeout ( timeoutCallBack ( userId ) ) ;
 sseEmitterMap .put ( userId , sseEmitter ) ;
 count .getAndIncrement ( ) ;
 return sseEmitter ;
 } catch ( Exception e ) {
 log.info ( "Exception in creating a new sse connection, current user: {}" , userId ) ;
 }
 return null ;
 }

 /**
 * Send a message to a specified user
 *
 * @date: 2022/7/12 14:51
 * @auther: Public account: Programmer Xiaofu
 */
 public static void sendMessage ( String userId , String message ) {

 if ( sseEmitterMap .containsKey ( userId ) ) {
 try {
 sseEmitterMap .get ( userId ) .send ( message ) ;
 } catch ( IOException e ) {
 log .error ( "User [{}] push exception: {}" , userId , e .getMessage ( ) ) ;
 removeUser ( userId ) ;
 }
 }
 }

We simulate the server pushing messages and see if the client receives the messages, which is consistent with our expectations.

Note: SSE does not support IE browser, but it has good compatibility with other mainstream browsers.

MQTT

What is MQTT protocol?

MQTT (Message Queue Telemetry Transport): A lightweight communication protocol based on the publish/subscribe model. It obtains messages by subscribing to corresponding topics. It is a standard transmission protocol in the Internet of Things.

The protocol separates the publisher and subscriber of the message, so it can provide reliable message services for remotely connected devices in an unreliable network environment. Its usage is somewhat similar to traditional MQ.

The TCP protocol is located at the transport layer, and the MQTT protocol is located at the application layer. The MQTT protocol is built on the TCP/IP protocol, which means that the MQTT protocol can be used wherever the TCP/IP protocol stack is supported.

Why use MQTT protocol?

Why is the MQTT protocol so popular in the Internet of Things (IOT) instead of other protocols, such as the more familiar HTTP protocol?

• First of all, HTTP is a synchronous protocol. After the client makes a request, it needs to wait for the server's response. In the Internet of Things (IOT) environment, devices are subject to environmental influences, such as low bandwidth, high network latency, unstable network communication, etc. Obviously, asynchronous message protocols are more suitable for IOT applications.
• HTTP is one-way, and the client must initiate a connection if it wants to get a message. In Internet of Things (IOT) applications, devices or sensors are often clients, which means they cannot passively receive commands from the network.
• Usually a command or message needs to be sent to all devices on the network. It is not only difficult but also very costly to implement such a function with HTTP.

I will not go into the specific introduction and practice of the MQTT protocol here. You can refer to my previous two articles, which are also very detailed.

Introduction to MQTT protocol

I didn't expect that using springboot + rabbitmq for smart home would be so simple

MQTT implements message push

Unread messages (little red dots), front-end and RabbitMQ real-time message push practice, very simple~

Websocket

Websocket should be a way of implementing message push that everyone is familiar with. When we talked about SSE above, we also compared it with websocket.

WebSocket is a full-duplex communication protocol on a TCP connection, establishing a communication channel between the client and the server. The browser and the server only need one handshake to directly create a persistent connection between the two and perform two-way data transmission.

The picture comes from the Internet

Springboot integrates websocket and first introduces websocket-related toolkits, which has additional development costs compared to SSE.

 <!-- Introducing websocket -->
 < dependency >
 < groupId > org .springframework .boot </ groupId >
 <artifactId> spring - boot - starter - websocket </artifactId>
 </ dependency >

The server uses the @ServerEndpoint annotation to mark the current class as a websocket server. The client can connect to the WebSocket server via ws://localhost:7777/webSocket/10086.

 @Component
 @Slf4j
 @ServerEndpoint ( "/websocket/{userId}" )
 public class WebSocketServer {
 // Connection session with a client, need to send data to the client through it
 private Session session ;
 private static final CopyOnWriteArraySet < WebSocketServer > webSockets = new CopyOnWriteArraySet <> ( ) ;
 // Used to store the number of online connections
 private static final Map < String , Session > sessionPool = new HashMap < String , Session > ( ) ;
 /**
 * Public account: Programmer Xiaofu
 * Link successfully called method
 */
 @OnOpen
 public void onOpen ( Session session , @PathParam ( value = "userId" ) String userId ) {
 try {
 this .session = session ;
 webSockets .add ( this ) ;
 sessionPool .put ( userId , session ) ;
 log.info ( "websocket message: There are new connections, the total is:" + webSockets.size ( ) ) ;
 } catch ( Exception e ) {
 }
 }
 /**
 * Public account: Programmer Xiaofu
 * The method called after receiving a client message
 */
 @OnMessage
 public void onMessage ( String message ) {
 log.info ( "websocket message: Received client message:" + message ) ;
 }
 /**
 * Public account: Programmer Xiaofu
 * This is a single message
 */
 public void sendOneMessage ( String userId , String message ) {
 Session session = sessionPool .get ( userId ) ;
 if ( session != null && session .isOpen ( ) ) {
 try {
 log.info ( "websocket message: single message:" + message ) ;
 session .getAsyncRemote ( ) .sendText ( message ) ;
 } catch ( Exception e ) {
 e .printStackTrace ( ) ;
 }
 }
 }
 }

The front end initializes and opens the WebSocket connection, monitors the connection status, and receives or sends data from the server.

 < script >
 var ws = new WebSocket ( 'ws://localhost:7777/webSocket/10086' ) ;
 // Get the connection status
 console .log ( 'ws connection status: ' + ws .readyState ) ;
 // Monitor whether the connection is successful
 ws .onopen = function ( ) {
 console .log ( 'ws connection status: ' + ws .readyState ) ;
 // If the connection is successful, send a data
 ws .send ( 'test1' ) ;
 }
 // Receive the information sent back by the server and process and display it
 ws .onmessage = function ( data ) {
 console .log ( 'Received message from server: ' ) ;
 console .log ( data ) ;
 // Close the WebSocket connection after completing the communication
 ws .close ( ) ;
 }
 // Listen for connection closing events
 ws .onclose = function ( ) {
 // Monitor the status of websocket during the whole process
 console .log ( 'ws connection status: ' + ws .readyState ) ;
 }
 // Listen for and handle error events
 ws .onerror = function ( error ) {
 console .log ( error ) ;
 }
 function sendMessage ( ) {
 var content = $ ( "#message" ) .val ( ) ;
 $ .ajax ( {
 url : '/socket/publish?userId=10086&message=' + content ,
 type : 'GET' ,
 data : { "id" : "7777" , "content" : content } ,
 success : function ( data ) {
 console .log ( data )
 }
 } )
 }
 </ script >

The page initializes and establishes a websocket connection, after which two-way communication can be carried out, and the effect is quite good

Custom push

We have provided the principles and code implementations of six solutions above, but in the actual business development process, we cannot blindly use them directly. We still need to choose the appropriate solution based on the characteristics of our own system business and the actual scenario.

The most direct way to push is to use a third-party push platform. After all, any demand that can be solved with money is not a problem. There is no need for complicated development and operation and maintenance. It can be used directly, saving time, effort and worry. GoEasy and Jiguang Push are both very good third-party service providers.

Generally, large companies have their own message push platforms. The web-based message we implemented this time is just a touchpoint on the platform. SMS, email, WeChat public accounts, mini programs and any channels that can reach users can be connected.

The picture comes from the Internet

The internal structure of the message push system is quite complex, with many modules, such as message content maintenance and review, target audience, reach filtering and interception (push rules, frequency, time period, quantity, blacklist and whitelist, keywords, etc.), and push failure compensation. The technology involves many scenarios with large data volumes and high concurrency. Therefore, our implementation today is just a small step in front of this huge system.

Github address

I have implemented all the cases mentioned in the article and put them on Github. If you find them useful, please star them!

Portal: https://github.com/chengxy-nds/Springboot-Notebook/tree/master/springboot-realtime-data