物聯網架構成長之路(49)-SpringBoot集成KafKa中間件

0.前言

　　今天(2020-02-24)是開工的第一天，來到公司后，服務器出現問題，網管正在處理。沒有服務器的后端，就像沒有武器的劍客。沒辦法進行開發，就看看資料學習一點技術。
　　疫情期間，雖然沒有上班，但是自己的物聯網平台還是在慢慢的優化中。下面這個圖是規划后的V2版本架構圖。
　　架構圖里面用到Kafka中間件，是作為數據流來處理。由於MQTT(EMQ)無法進行數據的持久化，所以需要引入Kafka來實現處理。EMQ用來保證通信的實時性和高效性。Kafka利用消息隊列特性用來進行非實時與離線處理。
　　比如架構圖所示，可以利用EMQ的Kafka插件或者訂閱MQTT根Topic的方式，把通信內容按照規則發往Kafka，作為生產者。而后面的支付服務，離線大數據處理服務，數據存儲服務等，作為消費者。

1. 利用Docker-Compose搭建kafka

　　docker-compose.yml

version: '3'
services:
    zookeeper:
        image: wurstmeister/zookeeper
        ports:
            - "2181:2181"
    kafka:
        image: "wurstmeister/kafka:2.12-2.4.0"
        ports:
            - "9092:9092"
        environment:
            KAFKA_ADVERTISED_HOST_NAME: 192.168.0.106
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
        volumes:
            - /var/run/docker.sock:/var/run/docker.sock
            - /root/workspace/kafka/data:/kafka
            - /etc/localtime:/etc/localtime
    kafka-manager:
        image: "sheepkiller/kafka-manager"
        restart: always
        container_name: kafka-manger
        ports:
            - "9091:9000"
        links:
            - zookeeper
            - kafka
        environment:
            ZK_HOSTS: zookeeper:2181
            KAFKA_BROKERS: kafka:9092
            KM_ARGS: -Djava.net.preferIPv4Stack=true
            KM_USERNAME: admin
            KM_PASSWORD: admin

　　下面是一些基礎操作

#進入容器
docker exec -it ${CONTAINER ID} /bin/bash 
#進入目錄
cd opt/kafka_2.11-0.10.1.1
#創建Topic
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic test
#查詢Topic
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

#運行一個生產者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>This is Message

#運行一個消費者
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

 

2. Kafka Manager界面
　　訪問192.168.0.106:9091，這個界面就是Kafka Manager管理界面，我們增加一個Cluster。然后可以查看對應的kafka信息

 

3. SpringBoot集成Kafka

3.1 pom.xml

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
        </dependency>

3.2 KafkaController.java

package com.wunaozai.demo.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(value="/kafka")
public class KafkaController {

    @Autowired
    private KafkaSender kafkaSender;
    
    @RequestMapping(value="/send")
    public String send(String msg) {
        boolean flag = kafkaSender.send(msg);
        return flag + "";
    }
}

3.3 KafkaMessage.java

package com.wunaozai.demo.kafka;

import java.sql.Timestamp;

public class KafkaMessage {
    
    private Long id;
    private String msg;
    private Timestamp ts;
    public Long getId() {
        return id;
    }
    
    public void setId(Long id) {
        this.id = id;
    }
    
    public String getMsg() {
        return msg;
    }
    
    public void setMsg(String msg) {
        this.msg = msg;
    }
    
    public Timestamp getTs() {
        return ts;
    }
    
    public void setTs(Timestamp ts) {
        this.ts = ts;
    }
    
}

3.4 KafkaReceiver.java

package com.wunaozai.demo.kafka;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaReceiver {

    private static final Logger log = LoggerFactory.getLogger(KafkaReceiver.class);

    @KafkaListener(topics= {"iot"})
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> message = Optional.ofNullable(record.value());
        if(message.isPresent()) {
            Object msg = message.get();
            log.info("record : " + record);
            log.info("message : " + msg);
        }
    }
}

3.5 KafkaSender.java

package com.wunaozai.demo.kafka;

import java.sql.Timestamp;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

@Component
public class KafkaSender {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private Gson gson = new GsonBuilder().create();
    
    public boolean send(String msg) {
        KafkaMessage message = new KafkaMessage();
        message.setId(System.currentTimeMillis());
        message.setTs(new Timestamp(System.currentTimeMillis()));
        message.setMsg(msg); 
        kafkaTemplate.send("iot", gson.toJson(message));
        return true;
    }
}

3.6 application.properties

#指定kafka 代理地址，可多個
spring.kafka.bootstrap-servers=192.168.0.106:9092

#provider
spring.kafka.producer.retries=0
#每次批量發送消息的數量
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
#指定消息key和消息體body的編解碼方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#consumer
# 指定默認消費者group id
spring.kafka.consumer.group-id=0
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100
# 指定消息key和消息體的編解碼方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

 運行效果圖

　　經過測試，通過java方式與console方式都是可以互相通信。

　　假設在KafkaReceiver.java 第23行增加 Thread.sleep(10000); 用來模擬實際業務延時。類似搶票或者秒殺的應用場景。突然高峰，然后把所有訂單數據都放到Kafka，然后在慢慢消費。生成實際業務訂單，再通知用戶付款。

　　按照架構設計，平台部分會訂閱EMQ的Topic，過濾部分數據或者完整數據報都發往Kafka，然后讓Kafka后面的消費者根據需要自己進行消費。比如對所有 iot/product-uuid/device-uuid/device/+/property 所有屬性日志相關的消息都通過Kafka的property這個Topic發送。消費者訂閱property后，就可以消費。至於消費后可以存入influxdb進行持久化，也可以預處理后顯示。MQTT設計中還有一類是 iot/product-uuid/device-uuid/device/+/event 事件類。這些相關的消息，會發往Kafka的event主題。然后由后面的消費者來消費event。進而進行報警等處理。

參考資料：
　　http://kafka.apachecn.org/documentation.html#operations
　　https://hub.docker.com/r/wurstmeister/kafka
　　https://mp.weixin.qq.com/s?__biz=MzU2NDg0OTgyMA==&mid=2247484570&idx=1&sn=1ad1c96bc7d47b88e976cbd045baf7d7

本文地址：https://www.cnblogs.com/wunaozai/p/12358247.html
本系列目錄： https://www.cnblogs.com/wunaozai/p/8067577.html
個人主頁：https://www.wunaozai.com/