下面就是quarkus具体的使用了，Quarkus 官方提供了两种快速初始化项目的方式：

• maven脚手架快速搭建项目
• Quarkus start coding网站快速搭建项目

两种方式都支持生成maven、gradle项目

maven脚手架快速搭建项目

使用Quarkus Maven插件可以创建一个简单的项目，该项目可以随时部署，包含以下内容：·

• 一个包含最小Quarkus依赖的pom.xml文件
• 一个简单的JAX-RS资源
• 一个JAX-RS资源的测试
• 原生测试
• 用于生成容器的Dockerfile
• 一个空的配置文件

1

mvn io.quarkus:quarkus-maven-plugin:2.15.3.Final:create -DprojectGroupId=com.liucongblog.quarkus -DprojectArtifactId=quarkus-start -DclassName="com.liucongblog.quarkus.QuarkusStart" -Dpath="/hello"

将生成的项目导入IDEA就可以直接加载成maven项目并启动了。

Quarkus start coding网站快速搭建项目

登录 https://code.quarkus.io/ 即可快速搭建一个quarkus项目如下图所示，只需要填写group name， artifactId， 构建工具，选择需要的依赖包，点击Generate your application即可生成项目：

将生成的项目导入IDEA:

可以看到生成的项目直接就可以使用，甚至连dockerfile都帮我们准备好了，确实很方便，
也贴心的准备了项目的测试和集成测试示例。

结论

quarkus的脚手架搭建初始化项目确实很快，基本跟spring的intializer一样，甚至做得更多，包括项目的构建Dockerfile也准备好了。
使用Maven脚手架和Web都可以快速搭建，还是使用start coding的网页更方便。

Spring已经发布很久了，从Spring改变SOA架构到现在已经快二十年了，现在都在容器化、云原生，一些旧的理念已经不太适应这个容器化的环境，所以这次要介绍一个云原生的微服务框架Quarkus。

Quarkus是什么

Quarkus是由Red Hat公司于2018年开始研发的一款面向云原生的Java开发框架，旨在使 Java 成为Kubernetes 和无服务环境中的领先平台。Quarkus被称为“超音速、亚原子”（supersonic，subatomic）的Java框架。

Quarkus的官网是 https://quarkus.io/， 在github的仓库是：https://github.com/quarkusio/quarkus

记得19年刚刚尝试Quarkus的时候还是0.xx的版本，现在已经2.xx的版本了，Quarkus确实发展很快，各方面的功能已经越来越完善了，红帽作为开源社区的领导者也确实做的很好。

Quarkus的特点

• 云原生：支持通过GraalVM Native-Image将Java应用打包成本地二进制镜像，减少内存使用、缩短应用启动时间。这个原生镜像的能力确实很强，打包的镜像特别小，启动快，在服务切换的时候特别有用。

• 低使用成本：与常见的Java标准、框架等协同工作，如：Spring、Hibernate、Netty、RestEasy等，无需学习新的标准和规范. 严格来说Quarkus使用的是CDI的Bean标准，跟Spring自己定制的标准不是同一个，各种注解都是有差异的，但是差异不算太大。

• 高开发效率：代码热更新，无需重启即可查看代码改动的结果（dev环境下），支持单元测试、本地debug等。 代码热更新？ 其实这个springboot的 devtools也可以实现，并不是只有这家才有的。。

• 同时支持命令式和响应式代码。 响应式框架Quarkus使用的是mutiny这个框架，spring生态也有对应的Reactor 框架，在web端的实现就是Webflux。

• 支持同时运行在GraalVM和HotSpot两种虚拟机上。其实虚拟机这个Spring现在也支持了。

我觉得最主要的特性还是云原生的支持吧，启动速度快，支持打包原生镜像：
比如quarkus官网这个性能对比图：

可以看到quarkus打包的镜像rest接口的只有12MB，启动速度只需要0.016， 比起普通的java应用真是快多了。这也是quarkus作为云原生框架能跟go 抗衡的一点吧。

为什么用Quarkus

为什么用Quarkus？ 可能还是处于技术人的好奇心吧，其实各种框架性的功能Spring、springboot、springcloud都提供了，但Spring已经越来越臃肿了，各种依赖，各种限制，越来越不适应云原生的应用环境了。有时间我们只是一个很简单的CRUD应用，使用spring生态开发出来，开发时间又长，镜像又大， 启动时间又慢。 而Quarkus正是为了解决我们在容器化环境开发微服务的问题而诞生的，启动ms级，镜像只有10多M， 这样我们的微服务容器更新版本的时候只需要几秒就可以拉取完镜像，并且启动快速，迅速可以替换掉旧的应用。真是太方便了， 非常值得一学！

引言

最近感觉笔记本风扇声音太大，感觉用太久了，需要清灰，顺带也可以换了硅脂了，本文主要记录下清灰和换硅脂的过程。

经过

1. 拆机前,准备好笔记本和相关工具，电动螺丝刀、硅脂:

2. 拆完地面之后:

3. 拆下散热铜管看到CPU,不愧是台式机的CPU！:

4. 底面电池壳无法完全拆卸，有一个固定螺丝在笔记本的键盘面上，如图红点所示：

5. 将风扇和CPU清理干净，去除旧的硅脂:

6. 在CPU和散热铜管涂上新的硅脂:

7. 大功告成，将底盖安装上，拧好螺丝:

8. 开机使用鲁大师测量指标，感觉温度没有下降很多，可能是使用的硅脂不行:

收获

• 太久没折腾电脑硬件了，拆机都不太熟练了，有些螺丝多拆了，我只需要清灰和换硅脂其实只需要拆底面的4颗螺丝和铜管的6颗螺丝的。 哎！ 果然都是经验活！
• 才发现原来我的笔记本还有一个SSD卡槽，这笔记本有2个M2的硬盘槽，这下我可以直接加装一个新的SSD硬盘就好了，正好我现在C盘空间不够。

缘由

起因是想好好利用自己买来的服务器做一些辅助功能，单纯的linux server已经不够了，需要使用 windows server。以下仅记录我使用过程中踩坑碰到的问题。

Win10⽆法完成操作,因为⽂件包含病毒或潜在的垃圾软件

这个问题通常是因为杀软对于一些恶意程序的判别导致的，会在恶意程序解压后直接删除，或者报上面这个错误。

• 解决办法⼀：按Windows+R，输⼊gpedit.msc，打开组策略编辑器，通过计算机配置>>管理模板>>Windows组件>>Windows Defender防病毒程序>>打开右边的允许反恶意软件服务始终保持运⾏状态，设置为禁⽤。
  
• 解决办法⼆：
  1. 单击左下⾓windows图标，找到「设置」，然后找到打开「更新和安全」
  2. 点击菜单中的「Windows安全中⼼」，再点右侧上⽅「打开Windows安全中⼼」
  3. 找到并打开「”病毒和威胁防护”设置」，点击下⾯的「管理设置」
  4. 然后将「实时保护」关闭，即可PS：如果只想针对具体⽂件、⽂件夹、⽂件类型、进程进⾏排除，可以在下⽅找到「添加或删除排除项」进⾏设置即可。

一般来说找到windows defender 的配置关闭即可，但是由于是windows server，该项的配置跟个人版不一样，所以我是采用解决方法一解决的。

KERNELBASE.dll 0x01d87884f5f53018

当程序能正常解压不被杀掉之后，启动的时候还会有各种各样的问题，我这里碰到的是windows的.net 调用问题，具体错误信息如下：

1
2
3
4
5
6
7
8
9
10
11

错误应用程序名称: xxx.exe，版本: 2022.6.2.1，时间戳: 0x62984a14
错误模块名称: KERNELBASE.dll，版本: 10.0.14393.4886，时间戳: 0x61d521be
异常代码: 0xe0434352
错误偏移量: 0x000dc732
错误进程 ID: 0x16a4
错误应用程序启动时间: 0x01d87884f5f53018
错误应用程序路径: C:\Users\Administrator\Desktop\2022.6.2.1\xxx.exe
错误模块路径: C:\Windows\System32\KERNELBASE.dll
报告 ID: b0d1b0bd-9bdf-4ae4-97f6-ae9dfb546d8a
错误程序包全名: 
错误程序包相对应用程序 ID: 

通过搜索确认是windows的依赖库问题，然后查询本地的依赖库发现已经装了很多依赖库版本，通过咨询他人确认需要的vc2015_86.exe这个依赖库才能正常运行，所以以后使用windows程序都需要了解这些依赖的问题。

本地依赖库情况:

前言

被同事安利了，最近开始学习函数式语言clojure了，clojure是一种LISP峰哥的语言，运行在JVM上。Clojure的两个特点就是不可变数据和STM（软件事务存储:Software Transactinal Memory）。尝试了一下Clojure，不得不说确实对于思维方式有很大的改变，对于编程的心智负担也降低很多。

环境搭建

前言

之前使用rancher的命令快速构建了一个kubernetes集群，但是对于kubernetes内部组件的构建是缺失的，这次准备用kubernetes官方的工具kubeadm搭建一套kubernetes集群。
主要分为以下步骤：

• 机器准备和环境准备
• 安装kubeadm
• 使用kubeadm创建集群
• 使用kubeadm API定制组件
• 高可用了解

机器准备和环境准备

这里准备了5台2核4G的机器，然后用1台作为主节点，其他4台作为工作节点。

准备开始

• 一台兼容的 Linux 主机。Kubernetes 项目为基于 Debian 和 Red Hat 的 Linux 发行版以及一些不提供包管理器的发行版提供通用的指令。我这里使用的是centos7.9的5台机器
• 每台机器 2 GB 或更多的 RAM（如果少于这个数字将会影响你应用的运行内存）。每台机器2核4G符合要求
• CPU 2 核心及以上。每台机器2核4G符合要求
• 集群中的所有机器的网络彼此均能相互连接（公网和内网都可以）。这些机器都在一个vpc内网。
• 节点之中不可以有重复的主机名、MAC 地址或 product_uuid。使用ifconfig -a和 sudo cat /sys/class/dmi/id/product_uuid q确认了MAC地址和product uuid是唯一的。
• 开启机器上的某些端口。主要是6443端口，使用 nc -l 6443 -v 监听6443端口，然后另一台机器 nc x.x.x.x 6443命令确认网络连通性，我这里直接在云主机后台修改安全组规则即可生效。
• 禁用交换分区。为了保证 kubelet 正常工作，必须禁用交换分区。查看交换分区状态swapon --show , 临时禁用swapoff -a,永久禁用编辑 ​​/etc/fstab​​ 文件，注释 swap 类型的行。

容器运行时

为了在 Pod 中运行容器，Kubernetes 使用 容器运行时（Container Runtime）。
默认情况下，Kubernetes 使用 容器运行时接口（Container Runtime Interface，CRI） 来与你所选择的容器运行时交互。
如果你不指定运行时，kubeadm 会自动尝试通过扫描已知的端点列表来检测已安装的容器运行时。
在linux环境下需要以下的容器运行时套接字：

这里我是选择了docker ce ,但也改了containerd参数：

1
2
3
4
5
6
7
8
9
10
11
12

sudo yum remove docker   docker-client   docker-client-latest  docker-common   docker-latest   docker-latest-logrotate  docker-logrotate   docker-engine
sudo yum install -y yum-utils
sudo yum-config-manager   --add-repo https://download.docker.com/linux/centos/docker-ce.repo

sudo yum -y install docker-ce docker-ce-cli containerd.io docker-compose-plugin
##containerd  systemd
mkdir /etc/containerd
containerd config default > /etc/containerd/config.toml
sed -i 's/SystemdCgroup = false/SystemdCgroup = true/g'  /etc/containerd/config.toml
systemctl enable containerd --now
systemctl restart containerd --now

为了让 Linux 节点的 iptables 能够正确查看桥接流量，请确认 sysctl 配置中的 net.bridge.bridge-nf-call-iptables 设置为 1。例如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

cat <<EOF | sudo tee /etc/modules-load.d/k8s.conf
overlay
br_netfilter
EOF

sudo modprobe overlay
sudo modprobe br_netfilter

# 设置所需的 sysctl 参数，参数在重新启动后保持不变
cat <<EOF | sudo tee /etc/sysctl.d/k8s.conf
net.bridge.bridge-nf-call-iptables  = 1
net.bridge.bridge-nf-call-ip6tables = 1
net.ipv4.ip_forward                 = 1
EOF

# 应用 sysctl 参数而不重新启动
sudo sysctl --system

安装kubeadm、kubelet 和 kubectl

需要在每台机器上安装以下的软件包：

• kubeadm：用来初始化集群的指令。
• kubelet：在集群中的每个节点上用来启动 Pod 和容器等。
• kubectl：用来与集群通信的命令行工具。
  使用以下命令安装以上三个工具（这里使用阿里云的镜像站，谷歌官方的因为网络原因无法正常连接）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

cat <<EOF | sudo tee /etc/yum.repos.d/kubernetes.repo
[kubernetes]
name=Kubernetes
baseurl=https://mirrors.aliyun.com/kubernetes/yum/repos/kubernetes-el7-\$basearch
enabled=1
gpgcheck=1
gpgkey=https://mirrors.aliyun.com/kubernetes/yum/doc/rpm-package-key.gpg
exclude=kubelet kubeadm kubectl
EOF

# 将 SELinux 设置为 permissive 模式（相当于将其禁用）
sudo setenforce 0
sudo sed -i 's/^SELINUX=enforcing$/SELINUX=permissive/' /etc/selinux/config

sudo yum install -y kubelet kubeadm kubectl --disableexcludes=kubernetes

sudo systemctl enable --now kubelet

对于kubeadm安装的问题排查可以使用下面的链接查询 troubleshooting

使用kubeadm创建集群

下载集群镜像（可忽略）

由于国内网络原因，我这里准备了一个镜像拉取脚本用于拉取阿里云镜像替换默认谷歌镜像：

1
2
3
4
5
6
7
8
9

#!/bin/bash
url=registry.cn-hangzhou.aliyuncs.com/google_containers
version=v1.25.4
images=(`kubeadm config images list --kubernetes-version=$version|awk -F '/' '{print $2}'`)
for imagename in ${images[@]} ; do
  ctr images    pull $url/$imagename
  ctr images tag $url/$imagename registry.k8s.io/$imagename
  
done

将脚本复制到机器并授权执行：

1
2
3

chmod u+x image.sh
./image.sh
ctr images

执行后如图所示：
在每台机器都如此操作一次。

初始化master

通过ifconfig -a 命令确定master节点的内网ip和公网ip，然后使用以下命令初始化master节点，我这里使用的是flannel网络，所以是10.244.0.0：

1

kubeadm init  --image-repository=registry.aliyuncs.com/google_containers --apiserver-advertise-address x.x.x.x --apiserver-cert-extra-sans x.x.x.x --pod-network-cidr 10.244.0.0/16   --v=5

apiserver-advertise-address指定master的interface，pod-network-cidr指定Pod网络的范围
apiserver-cert-extra-sans strings用于 API Server 服务证书的可选附加主题备用名称（SAN）。可以是 IP 地址和 DNS 名称，这里将公网ip加入证书信息，以便公网连接api server。
中间还是有pause镜像问题，单独拉取了pause镜像：

1
2
3

ctr -n k8s.io image pull registry.cn-hangzhou.aliyuncs.com/google_containers/pause:3.6
ctr -n k8s.io image tag registry.cn-hangzhou.aliyuncs.com/google_containers/pause:3.6 k8s.gcr.io/pause:3.6
ctr -n k8s.io image tag registry.cn-hangzhou.aliyuncs.com/google_containers/pause:3.6 registry.k8s.io/pause:3.6

同时centos有libseccomp版本问题，需要手工安装高版本:

1
2

wget http://rpmfind.net/linux/centos/8-stream/BaseOS/x86_64/os/Packages/libseccomp-2.5.1-1.el8.x86_64.rpm
rpm -ivh libseccomp-2.5.1-1.el8.x86_64.rpm 

安装完毕截图如下:

接着按照提示执行复制kube config:

1
2
3
4

mkdir -p $HOME/.kube
sudo cp -i /etc/kubernetes/admin.conf $HOME/.kube/config
sudo chown $(id -u):$(id -g) $HOME/.kube/config

安装pod网络

初始化master之后就是安装集群网络了，我这里使用的是flannel网络：

1

kubectl apply -f https://raw.githubusercontent.com/coreos/flannel/master/Documentation/kube-flannel.yml

Node节点加入集群

接着便是安装kubeadm的提示安装node节点

1
2

kubeadm join x.x.x.x:6443 --token xxxx   --discovery-token-ca-cert-hash sha256:xxxxxx

执行完毕如图:

最终在master 节点执行kubectl get nodes结果如图，一个主节点四个工作节点的kubernetes集群就安装好了:

前言

实际使用kubernetes做微服务开发也有一段时间了，发觉自己对于kubernetes一些概念还是理解不够透彻，都说温故而知新，所以准备重新开始学习kubernetes。
首先当然是需要部署安装一个kubernetes环境了。

kubernetes介绍

Kubernetes 是由google，红帽，IBM发起的开源的容器编排项目，是一个开源的容器编排引擎，用来对容器化应用进行自动化部署、 扩缩和管理。该项目托管在 CNCF。

开发工具

• kubectl: Kubernetes 命令行工具，可以对 Kubernetes 集群运行命令,可以使用 kubectl 来部署应用、监测和管理集群资源以及查看日志,可安装在OSX,linux,windows 平台。开发友好的工具，基本开发调试部署kubernetes应用都会使用这个工具
• kind： 能够在本地计算机上运行 Kubernetes。 kind 要求安装并配置好 Docker
• minikube: 与 kind 类似，minikube 是一个工具， 能在本地运行 Kubernetes。 minikube 在本地的个人计算机（包括 Windows、macOS 和 Linux PC）运行一个单节点的 Kubernetes 集群，以便尝试 Kubernetes 或者开展每天的开发工作。
• kubeadm: kubeadm 可以创建和管理 Kubernetes 集群。 该工具能够执行必要的动作并用一种用户友好的方式启动一个可用的、安全的集群。

k8s企业发行版

• EKS(Amazon Elastic Kubernetes Service): EKS是一项托管服务，于 2018 年 6 月公开提供，用于在 AWS 上运行 Kubernetes。EKS是云上的收费平台。

• AKS(Azure Kubernetes Service): Azure Kubernetes 服务 (AKS) 是一种托管 Kubernetes 解决方案，由 Microsoft 于 2018 年推出。AKS是云上的收费平台。

• GKE(Google Kubernetes Engine): Kubernetes 本身是作为 Google 的内部项目开始的，因此他们在 2014 年率先提供托管 Kubernetes 解决方案，即 Google Kubernetes Engine (GKE)，这是有道理的。GKE是云上的收费平台。

• Openshift: 红帽提供的kubernetes发行版，openshift原先作为paas平台提供服务，后续改换为kubernetes内核的云平台。Openshift是开源的平台，由红帽提供企业支持。

• Rancher: 2014年由CoreOS提出的容器化解决方案的paas平台，后续2016年切换平台内核到kubernetes，提供完全免费和开源的kubernetes集群管理服务。

如果只是个人本地学习kubernetes概念，那么使用minikube， kind安装本地k8s集群已经够了，也可以使用kubeadm安装kubernetes体验完整的kubernetes，但是个人还是希望多学习一下企业kubernetes集群的功能，所以我选择使用rancher提供的kubernetes集群体验安装和部署。

使用rancher安装kubernetes集群

rancher支持在多种云环境下部署，同时也支持虚拟机部署，这里我选择手工在虚拟机部署

• 在 AWS 上快速部署
• 在 DigitalOcean 上快速部署
• 在 Azure 上快速部署
• 在 GCP 上快速部署
• 使用 Vagrant 进行快速部署
• 虚拟机手动部署

安装准备

为了部署rancher集群，我准备了四台腾讯云的轻量服务器集群作为k8s集群的master和node节点。

安装步骤:

• 运行 Linux 主机
• 安装 Rancher 2.x
• 登录 Rancher 2.x 并配置初始设置
• 创建业务集群

运行 Linux 主机

需要运行一台 Linux 主机，Rancher 支持的 Linux 主机类型包括：

• 云端虚拟机（如 Amazon ECS、阿里云 ECS、腾讯云 CVM、华为云 ECS 等）
• 本地数据中心的虚拟机
• 裸金属服务器（Bare Metal Server）

这里我使用腾讯云轻量服务器:

安装 Rancher 2.x

登录到主机，然后使用 shell 安装 Rancher。
执行以下命令：

1

sudo docker run -d --restart=unless-stopped -p 80:80 -p 443:443 --privileged rancher/rancher

登录准备好的域名查看rancher dashboard:

登录 Rancher 2.x 并配置初始设置

登录 Rancher登录以后，需要完成一些一次性的配置。打开浏览器，输入主机的 IP 地址：https://，我这里是 https://x.x.x.x
首次登录时，请按照页面提示设置登录密码和server url,设置 Rancher Server URL。URL 既可以是一个 IP 地址，也可以是一个主机名称。需要确保在集群内添加的每个节点都可以连接到这个 URL。如果使用的是主机名称，请保证主机名称可以被节点的 DNS 服务器成功解析。:

创建业务集群

完成安装和登录 Rancher 的步骤之后，使用以下步骤，在 Rancher 中创建第一个 Kubernetes 集群。可以使用自定义集群选项，使用的任意 Linux 主机（云主机、虚拟机或裸金属服务器）创建集群。

• 访问集群页面，单击添加集群。

• 选择已有节点选项。
  

• 输入集群名称，跳过集群角色和集群选项，单击下一步:
  

• 勾选主机选项 - 角色选择中的所有角色： Etcd、 Control 和 Worker。可选： Rancher 会自动探查用于 Rancher 通信和集群通信的 IP 地址。您可以通过主机选项 > 显示高级选项中的公网地址和内网地址指定 IP 地址。跳过主机标签参数，因为对快速入门来说，这部分的参数不太重要。这里我选择一台2C4G的机器作为etcd和controller节点，2台作为worker节点
  

• 复制代码框中的命令。登录 Linux 主机，打开命令行工具，粘贴命令，单击回车键运命令。运行完成后，回到 Rancher 界面，单击完成。
  
  
  

结果

集群创建中，并分配给集群状态为Provisioning，Rancher 正在建立中:

只需要等集群的状态更新为Active后访问它。Active集群被分配了两个项目：Default项目：包含default命名空间;System项目:包含cattle-system、ingress-nginx、kube-public和kube-system等命名空间。
如下图，等待一段时间后，我们的rancher管理的k8s集群就建立好了:

好的，重学kubernetes的第一步完成了！ 准备个人的kubernetes集群了！

延迟测试

测试 从mqtt 客户端到 kafka写入消息时间以及kafka发送消息到设备端接收到消息的时间延迟.

测试场景 mqtt -> mq

准备

发送客户端使用简单的eclipse mqtt 客户端,简单地使用循环发送消息，每次发送前记录当前时间戳：

1
2
3
4
5
6
7
8

...
for (int i = 0; i < Integer.valueOf(args[2]); i++) {
     MqttMessage message = new MqttMessage(String.valueOf(i).getBytes(StandardCharsets.UTF_8));
     message.setQos(qos);
     System.out.println(" time:" + Instant.now().toEpochMilli() + ",Message publish:" + i);
     sampleClient.publish(args[1], message);
 }
 ...

这里使用main方法传入的第二个参数作为发送主题名，第一个参数作为mqtt服务器的地址，第三个参数作为循环次数,然后打包成 mqtt-producer.jar在服务器备用。

同时这里设置我们网桥程序内部缓存双向队列的空闲等待时间为4ms， 可以设置更低，这里设置4ms 避免占用太多CPU资源。

然后创建响应的kafka的主题，并设置主题的timestamp为kafka写入日志的时间： message.timestamp.type=LogAppendTime kafka 接收端就使用kafka官方提供的consumer 控制台打印时间戳到日志文件。

测试脚本

发送端直接使用准备的mqtt-producer.jar，执行命令将时间记录到日志文件:

1

java -jar mqtt-producer.jar tcp://192.168.0.2:1833 /tacos/pub/lf 1000000 >>pub_lf.txt &

在消息发送完成后使用kafka-console-consumer 打印时间戳到日志文件：

1

sh bin/kafka-console-consumer.sh  --topic _tacos_pub_lf  --bootstrap-server 192.168.0.5:9092,192.168.0.9:9092,192.168.0.13:9092   --property print.timestamp=true --from-beginning >>_tacos_pub_lf.txt &

测试结果

整理比对发送端和接收端的日志文件，比较是否存在数据差异，根据时间戳计算延迟（详情见附件result.xlsx表格mqtt to kafka）：

总结：mqtt发送消息到kafka 写入日志文件，平均耗时可以控制在2ms， 基本符合我们设置的内部缓存队列的超时实际4ms .

测试场景 mq -> mqtt

准备

发送客户端使用kafka提供的生产者性能测试工具进行压力测试，创建对应的主题，这里不需要单独设置主题的时间戳类型，使用默认的即生产者创建消息的实际作为时间戳。

接收端使用eclipse的客户端简单的打印接收到的信息和时间戳:

1

sampleClient.subscribe(args[1], 1, (topic1, message) -> System.out.println(" time:" + Instant.now().toEpochMilli() + ",Message arrived:" + message));

测试脚本

首先启动我们的mqtt消费端，将其控制台日志打印到日志文件:

1

java -jar mqtt-consumer.jar tcp://192.168.0.2:1833 /tacos/sub/l >>log.txt &

然后使用kafka压测工具发送消息:

1

sh bin/kafka-producer-perf-test.sh --topic _tacos_sub_le  --num-records 200000  --throughput 1000  --producer.config config/producer.properties --payload-file data.txt

发送完毕之后使用kafka消费者控制台记录时间戳到日志文件:

1

sh bin/kafka-console-consumer.sh  --topic _tacos_pub_le  --bootstrap-server 192.168.0.5:9092,192.168.0.9:9092,192.168.0.13:9092   --property print.timestamp=true --from-beginning >>_tacos_pub_le.txt &

测试结果

整理比对发送端和接收端的日志文件，比较是否存在数据差异，根据时间戳计算延迟（详情见附件result.xlsx表格kafka to mqtt）：

总结：kafka发送消息到mqtt设备端接收消息，平均耗时1.25ms， 基本符合我们的预期.