kafka和websocket實時資料推送

首頁>技術>江天明月2021-03-02 11:18

kafka和websocket實時資料推送

需求

已有Kafka服務，透過kafka服務資料（GPS）落地到本地磁碟（以文字檔案儲存）。現要根據echarts實現一個實時車輛的地圖。

分析前端實時展現：使用websocket技術，實現伺服器端資料推送到前端展現透過Java的kafka client端獲取資料，並且透過websock推送到前端。websocket簡介

websocket是HTML5開始提供的一種在單位TCP連線上進行全雙工通訊的協議。在websocket api中，瀏覽器和伺服器只需要做一次握手的動作，然後瀏覽器和伺服器之間就形成了一條快速通道。兩者之間就可以資料互相傳送。

開發伺服器端

  package com.ykkj.weiyi.socket;  import org.springframework.stereotype.Component;  import org.springframework.web.socket.server.standard.SpringConfigurator;  import javax.websocket.*;  import javax.websocket.server.ServerEndpoint;  import java.io.IOException;  import java.util.concurrent.CopyOnWriteArraySet;  /**   * @ServerEndpoint 註解是一個類層次的註解，它的功能主要是將目前的類定義成一個websocket伺服器端,   * 註解的值將被用於監聽使用者連線的終端訪問URL地址,客戶端可以透過這個URL來連線到WebSocket伺服器端   */  @ServerEndpoint(value = "/websocket")  public class CommodityServer {      //靜態變數，用來記錄當前線上連線數。應該把它設計成執行緒安全的。      private static int onlineCount = 0;      //concurrent包的執行緒安全Set，用來存放每個客戶端對應的MyWebSocket物件。若要實現服務端與單一客戶端通訊的話，可以使用Map來存放，其中Key可以為使用者標識      public static CopyOnWriteArraySet<CommodityServer> webSocketSet = new CopyOnWriteArraySet<CommodityServer>();      //與某個客戶端的連線會話，需要透過它來給客戶端傳送資料      private Session session;      /**       * 連線建立成功呼叫的方法       *       * @param session 可選的引數。session為與某個客戶端的連線會話，需要透過它來給客戶端傳送資料       */      @OnOpen      public void onOpen(Session session) {          this.session = session;          webSocketSet.add(this);     //加入set中          addOnlineCount();           //線上數加1          System.out.println("有新連線加入！當前線上人數為" + getOnlineCount());      }      /**       * 連線關閉呼叫的方法       */      @OnClose      public void onClose() {          webSocketSet.remove(this);  //從set中刪除          subOnlineCount();           //線上數減1          System.out.println("有一連線關閉！當前線上人數為" + getOnlineCount());      }      /**       * 收到客戶端訊息後呼叫的方法       *       * @param message 客戶端傳送過來的訊息       * @param session 可選的引數       */      @OnMessage      public void onMessage(String message, Session session) {          System.out.println("來自客戶端的訊息:" + message);          //群發訊息          for (CommodityServer item : webSocketSet) {              try {                  item.sendMessage(message);              } catch (IOException e) {                  e.printStackTrace();                  continue;              }          }      }      /**       * 發生錯誤時呼叫       *       * @param session       * @param error       */      @OnError      public void onError(Session session, Throwable error) {          System.out.println("發生錯誤");          error.printStackTrace();      }      /**       * 這個方法與上面幾個方法不一樣。沒有用註解，是根據自己需要新增的方法。       *       * @param message       * @throws IOException       */      public void sendMessage(String message) throws IOException {          this.session.getBasicRemote().sendText(message);          //this.session.getAsyncRemote().sendText(message);      }      public static synchronized int getOnlineCount() {          return onlineCount;      }      public static synchronized void addOnlineCount() {          CommodityServer.onlineCount++;      }      public static synchronized void subOnlineCount() {          CommodityServer.onlineCount--;      }  }複製程式碼

前端

<!DOCTYPE html><html><head>    <title>Java後端WebSocket的Tomcat實現</title></head><body>Welcome<br/><input id="text" type="text"/><button onclick="send()">傳送訊息</button><hr/><button onclick="closeWebSocket()">關閉WebSocket連線</button><hr/><div id="message"></div></body><script type="text/javascript">    var websocket = null;    //判斷當前瀏覽器是否支援WebSocket    if ('WebSocket' in window) {        websocket = new WebSocket("ws://localhost:8081/onepic/websocket");    }    else {        alert('當前瀏覽器 Not support websocket')    }    //連線發生錯誤的回撥方法    websocket.onerror = function () {        setMessageInnerHTML("WebSocket連線發生錯誤");    };    //連線成功建立的回撥方法    websocket.onopen = function () {        setMessageInnerHTML("WebSocket連線成功");    }    //接收到訊息的回撥方法    websocket.onmessage = function (event) {        setMessageInnerHTML(event.data);    }    //連線關閉的回撥方法    websocket.onclose = function () {        setMessageInnerHTML("WebSocket連線關閉");    }    //監聽視窗關閉事件，當視窗關閉時，主動去關閉websocket連線，防止連線還沒斷開就關閉視窗，server端會拋異常。    window.onbeforeunload = function () {        closeWebSocket();    }    //將訊息顯示在網頁上    function setMessageInnerHTML(innerHTML) {        document.getElementById('message').innerHTML += innerHTML + '<br/>';    }    //關閉WebSocket連線    function closeWebSocket() {        websocket.close();    }    //傳送訊息    function send() {        var message = document.getElementById('text').value;        websocket.send(message);    }</script></html>

測試注意點webSocketSet設定為全域性靜態變數，為其他類提供呼叫

public static CopyOnWriteArraySet<CommodityServer> webSocketSet = new CopyOnWriteArraySet<CommodityServer>();

服務端都是用註解實現的 @ServerEndpoint @OnOpen @OnClose @OnMessage @OnErrorTomcat7.0.47以上版本支援websocket1.0pom中新增jar支援

<dependency>    <groupId>javax</groupId>    <artifactId>javaee-api</artifactId>    <version>7.0</version>    <scope>provided</scope></dependency>

Kafka簡介

Kafka是一個分散式的、可分割槽的、可複製的訊息系統。

kafka client for java

package com.ykkj.weiyi.socket;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import java.io.IOException;import java.util.Arrays;import java.util.Properties;import static com.ykkj.weiyi.socket.CommodityServer.webSocketSet;public class ConsumerKafka extends Thread {    private KafkaConsumer<String, String> consumer;    private String topic = "test.topic";    public ConsumerKafka() {    }    @Override    public void run() {        //載入kafka消費者引數        Properties props = new Properties();        props.put("bootstrap.servers", "localhost:9092");        props.put("group.id", "ytna");        props.put("enable.auto.commit", "true");        props.put("auto.commit.interval.ms", "1000");        props.put("session.timeout.ms", "15000");        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        //建立消費者物件        consumer = new KafkaConsumer<String, String>(props);        consumer.subscribe(Arrays.asList(this.topic));        //死迴圈，持續消費kafka        while (true) {            try {                //消費資料，並設定超時時間                ConsumerRecords<String, String> records = consumer.poll(100);                //Consumer message                for (ConsumerRecord<String, String> record : records) {                    //Send message to every client                    for (CommodityServer webSocket : webSocketSet) {                        webSocket.sendMessage(record.value());                    }                }            } catch (IOException e) {                System.out.println(e.getMessage());                continue;            }        }    }    public void close() {        try {            consumer.close();        } catch (Exception e) {            System.out.println(e.getMessage());        }    }    //供測試用，若透過tomcat啟動需透過其他方法啟動執行緒    public static void main(String[] args) {        ConsumerKafka consumerKafka = new ConsumerKafka();        consumerKafka.start();    }}

注意topic 和 bootstrap.servers配置

呼叫類

  package com.ykkj.weiyi.socket;  public class RunThread {      public RunThread() {          ConsumerKafka kafka = new ConsumerKafka();          kafka.start();      }  }

web.xml 配置

<listener>  <listener-class>com.ykkj.weiyi.socket.RunThread</listener-class></listener>

測試注意點ConsumerKafka 需要在web.xml中配置監聽，不然在ConsumerKafka 類中不能獲取webSocketSet變數引用webSocketSet變數方式import static com.ykkj.weiyi.socket.CommodityServer.webSocketSet;注意topic 和 bootstrap.servers配置結束語

這裡只是做了技術驗證，還沒有真正實現。真實場景下，kafka獲取到的資料需要進行資料清洗，把不符合當前規範的資料清洗掉，並且按照前端展示需要的格式組裝資料。

參考網上資料

https://blog.csdn.net/lw_ghy/article/details/73252904

https://blog.csdn.net/liu857279611/article/details/70157012

∨ Web無程式碼線上開發的五大應用場景

熱門排行

劇多

kafka和websocket實時資料推送