【功能模块】关于kafka节点部署台数规划【操作步骤&问题现象】1、某局点kafka broker节点部署了4个，在开会讨论中，客户领导说kafka节点必须保持奇数，否则有台broker会浪费，而且leader选举时也会有问题2、但实际情况4台broker节点都进行了存储和使用，并没有出现客户说的问题，而且产品文档中也说明了，broker节点最少三台，并没有说必须保持奇数。只是zookeeper需要保持奇数。客户想让说明，问什么开源的需要保持奇数，但FusionInsight HD 没有这个要求，麻烦大佬帮忙解释下。【截图信息】【日志信息】（可选，上传日志内容或者附件）

传统的消息队列（（Kafka、RocketMQ等）经过多年打磨，在高性能、海量堆积、消息可靠性等诸多方面都已经做得非常极致，但在物联网场景中，往往需要面临着海量的消息传递，传统的消息队列表现的“力不从心”。IoT领域中，从应用服务器到嵌入式芯片，都需要传递事件消息，比如共享充电宝的开柜子、开灯指令从服务器发到设备、工业网关高频消息流等，在这些信息传递的过程中，队列最大意义在于让整个消息事件在不可控的环境因素变成一个平稳运行的系统，因为IoT设备时不时会由于故障或网络抖动会导致大量消息洪峰。阿里云AIoT作为物联网领域的引领者和创新者，在消息队列领域不断深耕与沉淀，为了让物联网从业者更进一步了解IoT场景队列，阿里云技术专家吕建文，整理了一份IoT队列的干货知识，与大家一同探讨一个适合于物联网系统的消息队列。一、IoT队列和普通队列的差异点1，上下行隔离拆分在IoT场景中，我们把需要队列分为两个场景，一个是上行队列，一个是下行队列。 拆分之后，可以隔离上下行链路，控制一个设备，比如支付成功要下发打开柜子等，上行出任何问题，千万不能影响到下行业务。另外，上下行两条链路的特点差异非常大。设备上行消息，并发量非常高，但很多场景下对于可靠性和时延要求低，而设备下行消息，并发量则比较低，但下行消息（一般是控制设备指令）要求到达成功率很高。2，支持设备级的海量topic传统队列的核心诉求是，不论堆积多少不影响它的性能。kafka的topic一多，原本消息顺序写文件优势就会导致一个broker要退化到随机写，失去优势，另外要zookeeper来协调这么多topic也是有局限，所以这些队列本身有提供一个外挂代理桥接器对外入口是多个设备topic，再桥接映射到少量的实际kafka topic，这方案有一定可行性，但做不到隔离效果，治标不治本。通过，图1和图2对比较明显，一个队列拥塞尽量减少对其它设备影响。我们需要的是“海量topic尽量相互隔离，并且不影响整体性能”，尽量做到设备A的消息堆积topic，不影响设备B。3，实时生成消息优先发送先举一个例子，一个快递柜业务的队列堆积，然后“此时此刻”在柜子旁边的用户死命的在旁边用手机点开柜子怎么也打不开（此时后端系统都恢复了），问题就是队列里面还有几十万条的消息，新来的消息需要排队, 等着之前的那些消息消费完，甭管这些消息还有没有用。  因此，实时生成消息优先发送，堆积的消息进入降级模式。二、IoT消息队列诞生1， IoT队列的设计思路设计目标是为了打造一个支持上下行隔离、实时优先、及海量topic的队列网关，设计原则如下：完全follow开源生态、和传统队列互补兼容保序降级，实时优先，堆积退化；仅实时消息相对有序。海量topic，多租户隔离连接、计算、存储分离2， 消息模式图片只是个片段，从这个模式可以看出来机制差别，大家都没有错，只是出发点不同。3， 连接、计算、存储分离broker不做连接，连接网关代理，broker只做流转分发，无状态+水平扩展；存储交给nosql DB，高吞吐写。4， 消息策略-推拉结合这个应该是队列的核心难点之一，和传统队列区分在于，我们考虑为平台化模式，独享资源过于昂贵。但带来问题是消费端不可控，所以使用结合模式，只有在消费者在线时会拉取堆积消息，而拉取是由AMQP队列网关来做，给到用户接口始终是推送过去的onMessage回调。broker不是直接让consumer来连接，而是把队列网关剥离出来，  这样会更灵活，甚至对于部分用户我们的queue可以切换到ons、kafka等实现。kafka、rocketmq做法是在连接时会分配给客户端一个broker接入地址。broker实时消息优先推送给consumer，失败才会落到queue ；这是一个完整事件，如果没有完成则不给producer commit。异步ACK5， 线性扩展-离线消息部分实时部分消息采用推方式，基本上不会成为瓶颈，消费不过来消息进入堆积模式。由于底层依赖存储已经帮我们解决核心存储的扩展，剩下主要问题点在于如何消除写入热点和消费热点，这样broker可以完全做到无状态。三，一个思考——如何解决海量topic问题？首先面对“大量”的问题一般都是考虑分区，单元化，分组等隔离和拆分，这里海量topic我们讨论针对一个单实例模式下如何尽可能做到更多topic，完全任意数量都能100%没问题肯定是不现实的。由于broker和存储已经隔离，broker和topic已经没有什么关系，或者说任何topic数据生成，broker做的事情就是写入和分发。海量topic，每个topic有限数量订阅：  topic和订阅者关系使用redis缓存或本地缓存，针对mqtt topic匹配有个topic tree的树算法，hivemq有实现版本。单个topic 海量订阅：  这个场景其实是组播和广播，我们不会考虑在队列本身上面去做这个事情，而是在上层封装广播组件来协调任务和批量发送。 四， 阿里云AIoT消息队列目前阿里云AIoT队列，也叫服务端订阅，意思就是用户用服务端订阅他们设备消息。为了降低接入成本，用户可以使用AMQP1.0协议接入，符合开源生态。 同时兼容传统队列和新队列，交给用户按场景来选择，用户即可选择使用kafka、mq，也可以选用iot队列，甚至组合模式，比如按消息特征规则来配置流转队列。阿里云AIoT的场景队列实践，在现有mq队列、kafka队列融合之外，加了种自有的实时优先队列实现，同时，加入了队列网关代理，既能让用户选择普通消息队列，也可以选择轻便的IoT消息队列。————————————————原文链接：https://blog.csdn.net/alitech2017/article/details/119563822

提示“您不在此活动范围。”为什么呀？不懂，瑟瑟发抖！

提示“您不在此活动范围。”为什么呀？不懂，瑟瑟发抖！

说明：本文参考https://bbs.huaweicloud.com/forum/thread-147823-1-1.html一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询：[root@1ocalhost ~]# cat /etc/os-release [root@1ocalhost ~]# uname -i2.软件版本   Kafka版本为：Kafka 2.7.03.JDK依赖   Kafka 是用Scala 语言开发的，运行在JVM上，因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境，需要自己自行安装。 这里安装JDK环境做简单说明，请开发者自行安装。[root@1ocalhost ~]# yum install java输入Y确认配置环境变量vim /etc/profile export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-1.h5.oe1.x86_64 export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/jre/lib/rt.jar export PATH=$PATH:$JAVA_HOME/bin环境变量生效source /etc/profile  查看JDK版本[root@1ocalhost ~]# java -version openjdk version "1.8.0_242" OpenJDK Runtime Environment (build 1.8.0_242-b08) OpenJDK 64-Bit Server VM (build 25.242-b08, mixed mode)二、详细安装步骤1.在/root目录创建data目录cd /root mkdir data2.进入data目录并下载Kafka 2.7.0 [root@1ocalhost ~]# cd data/ [root@1ocalhost data]# wget http://archive.apache.org/dist/kafka/2.7.0/kafka_2.12-2.7.0.tgz3.解压并进入到解压后的目录[root@1ocalhost data]# tar -zvxf kafka_2.12-2.7.0.tgz [root@1ocalhost data]# cd kafka_2.12-2.7.04.启动zookeeper服务和kafka服务，可以在命令的结尾加个&符号，这样服务就可以在后台运行   说明：单节点的 kafka 是不用修改配置文件，直接照官网的介绍就可以，见kafka官网入门[root@1ocalhost kafka_2.12-2.7.0]# /root/data/kafka_2.12-2.7.0/bin/zookeeper-server-start.sh /root/data/kafka_2.12-2.7.0/config/zookeeper.properties > zookeeper.log 2>&1 & [root@1ocalhost kafka_2.12-2.7.0]# /root/data/kafka_2.12-2.7.0/bin/kafka-server-start.sh /root/data/kafka_2.12-2.7.0/config/server.properties > kafkastart.log 2>&1 & 三、验证   kafka原理：Kafka集群将 Record 流存储在称为 Topic 的类别中，每个记录由一个键、一个值和一个时间戳组成。Kafka 中消息是以 Topic 进行分类的，生产者生产消息，消费者消费消息，面向的都是同一个 Topic。Topic 是逻辑上的概念，而 Partition 是物理上的概念，每个 Partition 对应于一个 log 文件，该 log 文件中存储的就是 Producer 生产的数据。Producer 生产的数据会不断追加到该 log 文件末端，且每条数据都有自己的 Offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个 Offset，以便出错恢复时，从上次的位置继续消费。1.创建一个叫"itren"的topic，它只有一个分区，一个副本：[root@1ocalhost kafka_2.12-2.7.0]# /root/data/kafka_2.12-2.7.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic itren2.发送消息，运行producer并在控制台中输一些消息，这些消息将被发送到服务端[root@1ocalhost kafka_2.12-2.7.0]# /root/data/kafka_2.12-2.7.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic itren >it520 >itren666 >byebye >3.在另一个终端开启consumer，可以读取到刚才发出的消息并输出[root@1ocalhost ~]# /root/data/kafka_2.12-2.7.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic itren it520 itren666 byebye

一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询：[root@ecs-2e3d ~]# cat /etc/os-release [root@ecs-2e3d ~]# uname -i2.软件版本   Kafka版本为：Kafka 2.5.03.JDK依赖   Kafka 是用Scala 语言开发的，运行在JVM上，因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境，需要自己自行安装。 这里安装JDK环境做简单说明，请开发者自行安装。[root@ecs-2e3d ~]# yum install java输入y 配置环境变量  JDK默认安装路径/usr/lib/jvm在/etc/profile文件末尾配置jdk路径vi /etc/profile# set java environment JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-1.h5.oe1.x86_64 PATH=$PATH:$JAVA_HOME/bin CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export JAVA_HOME CLASSPATH PATH保存关闭profile文件，执行如下命令生效source /etc/profile 查看JDK版本[root@ecs-2e3d ~]# java -version openjdk version "1.8.0_242" OpenJDK Runtime Environment (build 1.8.0_242-b08) OpenJDK 64-Bit Server VM (build 25.242-b08, mixed mode)二、详细安装步骤1.在/root创建data目录cd /root mkdir data2.进入data目录并下载Kafka 2.5.0 [root@ecs-2e3d ~]# cd data [root@ecs-2e3d data]# wget http://archive.apache.org/dist/kafka/2.5.0/kafka_2.12-2.5.0.tgz3.解压并进入到解压后的目录[root@ecs-2e3d data]# tar -zvxf kafka_2.12-2.5.0.tgz [root@ecs-2e3d data]# cd kafka_2.12-2.5.04.启动zookeeper服务和kafka服务，可以在命令的结尾加个&符号，这样服务就可以在后台运行   说明：单节点的 kafka 是不用修改配置文件，直接照官网的介绍就可以，见kafka官网入门[root@ecs-2e3d kafka_2.12-2.5.0]# /root/data/kafka_2.12-2.5.0/bin/zookeeper-server-start.sh /root/data/kafka_2.12-2.5.0/config/zookeeper.properties > zookeeper.log 2>&1 & [1] 7742 [root@ecs-2e3d kafka_2.12-2.5.0]# /root/data/kafka_2.12-2.5.0/bin/kafka-server-start.sh /root/data/kafka_2.12-2.5.0/config/server.properties > kafkastart.log 2>&1 & [2] 8115 [root@ecs-2e3d kafka_2.12-2.5.0]# 三、验证Kafka工作流程Kafka将消息按Topic进行分类，每条message由三个属性组成。          offset：表示message在当前Partition（分区）中的偏移量，是一个逻辑上的值，唯一确定了Partition中的一条message，可以简单的认为是一个id；           MessageSize：表示message内容data的大小；           data：message的具体内容；在整个kafka架构中，生产者和消费者采用发布和订阅的模式，生产者生产消息，消费者消费消息，它俩各司其职，并且都是面向topic的。（需要注意：topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据）Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，这样当出现故障并恢复后，可从这个offset位置继续进行消费，避免漏掉数据或者重复消费。1.创建一个叫"demo"的topic，它只有一个分区，一个副本：[root@ecs-2e3d kafka_2.12-2.5.0]# /root/data/kafka_2.12-2.5.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demo Created topic demo. [root@ecs-2e3d kafka_2.12-2.5.0]#2.发送消息，运行producer并在控制台中输一些消息，这些消息将被发送到服务端[root@ecs-2e3d kafka_2.12-2.5.0]# /root/data/kafka_2.12-2.5.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic demo >morning >nihao >goodday! >3.在另一个终端开启consumer，可以读取到刚才发出的消息并输出[root@ecs-2e3d kafka_2.12-2.5.0]# /root/data/kafka_2.12-2.5.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo morning nihao goodday!

一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询：[root@ecs-1d84 ~]# cat /etc/os-release [root@ecs-1d84 ~]# uname -i2.软件版本   Kafka版本为：Kafka 2.4.03.JDK依赖   Kafka 是用Scala 语言开发的，运行在JVM上，因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境，需要自己自行安装。 这里安装JDK环境做简单说明，请开发者自行安装。[root@ecs-1d84 ~]# yum install java输入y 配置环境变量  JDK默认安装路径/usr/lib/jvm在/etc/profile文件末尾配置jdk路径vi /etc/profile# set java environment JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-1.h5.oe1.x86_64 PATH=$PATH:$JAVA_HOME/bin CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export JAVA_HOME CLASSPATH PATH保存关闭profile文件，执行如下命令生效source /etc/profile查看JDK版本[root@ecs-1d84 ~]# java -version openjdk version "1.8.0_242" OpenJDK Runtime Environment (build 1.8.0_242-b08) OpenJDK 64-Bit Server VM (build 25.242-b08, mixed mode)二、详细安装步骤1.在/root创建data目录cd /root mkdir data2.进入data目录并下载Kafka 2.4.0 [root@ecs-1d84 ~]# cd data [root@ecs-1d84 data]# wget http://archive.apache.org/dist/kafka/2.4.0/kafka_2.11-2.4.0.tgz3.解压并进入到解压后的目录[root@ecs-1d84 data]# tar -zvxf kafka_2.11-2.4.0.tgz [root@ecs-1d84 data]# cd kafka_2.11-2.4.04.启动zookeeper服务和kafka服务，可以在命令的结尾加个&符号，这样服务就可以在后台运行   说明：单节点的 kafka 是不用修改配置文件，直接照官网的介绍就可以，见kafka官网入门[root@ecs-1d84 kafka_2.11-2.4.0]# /root/data/kafka_2.11-2.4.0/bin/zookeeper-server-start.sh /root/data/kafka_2.11-2.4.0/config/zookeeper.properties > zookeeper.log 2>&1 &[root@ecs-1d84 kafka_2.11-2.4.0]# /root/data/kafka_2.11-2.4.0/bin/kafka-server-start.sh /root/data/kafka_2.11-2.4.0/config/server.properties > kafkastart.log 2>&1 &三、验证   Kafka工作流程Kafka将消息按Topic进行分类，每条message由三个属性组成。          offset：表示message在当前Partition（分区）中的偏移量，是一个逻辑上的值，唯一确定了Partition中的一条message，可以简单的认为是一个id；           MessageSize：表示message内容data的大小；           data：message的具体内容；在整个kafka架构中，生产者和消费者采用发布和订阅的模式，生产者生产消息，消费者消费消息，它俩各司其职，并且都是面向topic的。（需要注意：topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据）Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，这样当出现故障并恢复后，可从这个offset位置继续进行消费，避免漏掉数据或者重复消费。1.创建一个叫"test"的topic，它只有一个分区，一个副本：[root@ecs-1d84 kafka_2.11-2.4.0]# /root/data/kafka_2.11-2.4.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test Created topic test. [root@ecs-1d84 kafka_2.11-2.4.0]# 2.发送消息，运行producer并在控制台中输一些消息，这些消息将被发送到服务端[root@ecs-1d84 kafka_2.11-2.4.0]# /root/data/kafka_2.11-2.4.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test >goodnight >20210818 >3.在另一个终端开启consumer，可以读取到刚才发出的消息并输出[root@ecs-1d84 kafka_2.11-2.4.0]# /root/data/kafka_2.11-2.4.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test goodnight 20210818现在所有完成，请指正。

在一般网络环境下，都存在一定的网络延迟、网络抖动，网络问题导致消息重复发送的情况是难以避免的，毕竟网络环境无法预知，因此MQ默认允许消息重复发送。是的，只要通过网络交换数据，就无法避免这个问题。秉承着打不过就加入的原则，解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？  RabbitMQ、RocketMQ、Kafka，都有可能会出现消息重复消费的问题。因为这问题通常不是 MQ 自己保证的，而是消费方自己来保证的。  比如说Kafka， 他实际上有个 offset 的概念（偏移量），就是每个消息写进去，都有一个 offset，代表消息的序号，然后 consumer 消费了数据之后，每隔一段时间（定时定期），会把自己消费过的消息的 offset 提交一下。代表我已经消费过了，就算消费者重启，Kafka也会让消费者继上次消费到的offset继续消费。场景示例：  kafka 中有一条数据：A、B，kafka给这条数据分一个 offset（偏移量），offset为： 1001、1002。消费者从 kafka 去消费的时候，也是按照这个顺序去消费。当消费者消费到 offset=1002 的这条数据（此时offset=1001还没消费完），刚提交 offset=1002 到 zookeeper，消费者进程就被重启了。此时消费过的数据 A 的 offset 还没有提交，kafka 也就不知道消费者已经消费了1001这条数据。那么重启之后，消费者会找 Kafka 把上次消费到的那个地方后面的数据继续传递过来。数据 A 再次被消费。

在一般网络环境下，都存在一定的网络延迟、网络抖动，网络问题导致消息重复发送的情况是难以避免的，毕竟网络环境无法预知，因此MQ默认允许消息重复发送。是的，只要通过网络交换数据，就无法避免这个问题。秉承着打不过就加入的原则，解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？  RabbitMQ、RocketMQ、Kafka，都有可能会出现消息重复消费的问题。因为这问题通常不是 MQ 自己保证的，而是消费方自己来保证的。  比如说Kafka， 他实际上有个 offset 的概念（偏移量），就是每个消息写进去，都有一个 offset，代表消息的序号，然后 consumer 消费了数据之后，每隔一段时间（定时定期），会把自己消费过的消息的 offset 提交一下。代表我已经消费过了，就算消费者重启，Kafka也会让消费者继上次消费到的offset继续消费。场景示例：  kafka 中有一条数据：A、B，kafka给这条数据分一个 offset（偏移量），offset为： 1001、1002。消费者从 kafka 去消费的时候，也是按照这个顺序去消费。当消费者消费到 offset=1002 的这条数据（此时offset=1001还没消费完），刚提交 offset=1002 到 zookeeper，消费者进程就被重启了。此时消费过的数据 A 的 offset 还没有提交，kafka 也就不知道消费者已经消费了1001这条数据。那么重启之后，消费者会找 Kafka 把上次消费到的那个地方后面的数据继续传递过来。数据 A 再次被消费。

一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询：[root@localhost ~]#cat /etc/os-release [root@localhost ~]# uname -i2.软件版本   Kafka版本为：Kafka 2.1.03.JDK依赖   Kafka 是用Scala 语言开发的，运行在JVM上，因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境，需要自己自行安装。 这里安装JDK环境做简单说明，请开发者自行安装，我这里安装的是 openjdk version "1.8.0_242"。 yum install java    输入 y  查看JDK版本[root@localhost ~]#java -version二、详细安装步骤1.在根目录创建data目录   mkdmkdir data ir data2.进入data目录并下载Kafka 2.1.0 [root@localhost /]# cd data [root@localhost data]# wget https://archive.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz3.解压并进入到解压后的目录[root@localhost data]# tar -zvxf kafka_2.11-2.1.0.tgz [root@localhost data]# cd kafka_2.11-2.1.0 4.启动zookeeper服务和kafka服务，可以在命令的结尾加个&符号，这样服务就可以在后台运行   说明：单节点的 kafka 是不用修改配置文件，直接照官网的介绍就可以，见kafka官网入门[root@localhost kafka_2.11-2.1.0]# ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &[root@localhost kafka_2.11-2.1.0]#./bin/kafka-server-start.sh ./config/server.properties &三、验证1.创建一个叫"kafkademo"的topic，它只有一个分区，一个副本：[root@localhost kafka_2.11-2.1.0]# ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafkademo2.发送消息，运行producer并在控制台中输一些消息，这些消息将被发送到服务端[root@localhost kafka_2.11-2.1.0]# ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafkademo3.在另一个终端开启consumer，可以读取到刚才发出的消息并输出[root@localhost kafka_2.11-2.1.0]# ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafkademo至此所有操作已完成，请指正。

【功能模块】按照官方的代码，提交flink消费kafka任务后，出现错误。在flink客户端使用的命令为bin/flink run -yt conf/ssl/ -ys 2 -m yarn-cluster -yjm 1024 -ytm 1024 -c org.mytest.stream.ReadFromKafka /opt/flink/flink.jar --topic topictest --bootstrap.servers $bs --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka【截图信息】【日志信息】（可选，上传日志内容或者附件）[root@node3 flink]# bin/flink run -yt conf/ssl/ -ys 2 -m yarn-cluster -yjm 1024 -ytm 1024 -c org.mytest.stream.ReadFromKafka /opt/flink/flink.jar --topic topictest --bootstrap.servers $bs --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafkaSLF4J: Class path contains multiple SLF4J bindings.SLF4J: Found binding in [jar:file:/opt/hadoopclient/Flink/flink/lib/logback-classic-1.2.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: Found binding in [jar:file:/opt/hadoopclient/HDFS/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.SLF4J: Actual binding is of type [ch.qos.logback.classic.util.ContextSelectorStaticBinder]Cluster started: Yarn cluster with application id application_1625541283910_0063Job has been submitted with JobID f50af8c85ed9c74e813f52c71231674fjava.util.concurrent.ExecutionException: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: f50af8c85ed9c74e813f52c71231674f)    at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)    at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:83)    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1651)    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1633)    at org.mytest.stream.ReadFromKafka.main(ReadFromKafka.java:21)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321)    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205)    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138)    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:700)    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:219)    at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:932)    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1005)    at java.security.AccessController.doPrivileged(Native Method)    at javax.security.auth.Subject.doAs(Subject.java:422)    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1737)    at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1005)Caused by: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: f50af8c85ed9c74e813f52c71231674f)    at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:112)    at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)    at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)    at org.apache.flink.client.program.rest.RestClusterClient.lambda$pollResourceAsync$21(RestClusterClient.java:565)    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)    at org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$8(FutureUtils.java:291)    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:575)    at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:943)    at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:456)    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)    at java.lang.Thread.run(Thread.java:748)Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.    at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)    at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:110)    ... 19 moreCaused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)    at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)    at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)    at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)    at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:484)    at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:380)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:279)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:194)    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at akka.actor.Actor$class.aroundReceive(Actor.scala:517)    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)    at akka.actor.ActorCell.invoke(ActorCell.scala:561)    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)    at akka.dispatch.Mailbox.run(Mailbox.scala:225)    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)    at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)    at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)    at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)    at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)Caused by: java.lang.NoSuchMethodError: org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.<init>(Lorg/slf4j/Logger;Lorg/apache/flink/streaming/connectors/kafka/internal/Handover;Ljava/util/Properties;Lorg/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue;Ljava/lang/String;JZLorg/apache/flink/metrics/MetricGroup;Lorg/apache/flink/metrics/MetricGroup;)V    at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.<init>(KafkaFetcher.java:109)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.createFetcher(FlinkKafkaConsumer.java:237)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:695)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:196)[root@node3 flink]# MRS版本：3.0.2flink版本：1.10.0

【功能模块】开发环境集群版本：6.5.1【操作步骤&问题现象】1、参考文档：Spring-kafka对接华为FusionInsight Kafka样例2、按照参考文档操作，测试时报没有方法错误【截图信息】pom文件：运行日志：【日志信息】（可选，上传日志内容或者附件）

##一、kafka简单介绍 kafka大家的第一印象是一个消息系统。但是kafka的官网的说法是： ## Apache Kafka® is a distributed streaming platform ##### kafka是一个分布式流处理平台，而流处理平台主要具备以下三种能力： 1.发布和订阅消息流，类似于消息队列或企业消息传递系统。 2.以容错的持久化方式储存消息流。 3.可以在产生消息流的时候，同时进行处理。 ##### 而kafka具备以下几个特性： 1.kafka作为一个集群可以运行在一个或者多个服务器上，这些服务器可以跨多个数据中心。 2.kafka集群存储的消息是按照topic(主题)进行分类的。 3.每个消息(也被称为记录)是由一个key，一个value和一个时间戳构成。 ##### kafka对外提供了四种核心API： 1.Producer API，允许应用程序发布消息到kafka集群上的1个或多个的topic。 2.Consumer API，允许应用程序订阅一个或多个topic，并处理这些topic的消息。 3.Streams API，允许应用程序充当一个流处理器，从1个或多个topic消费输入流，并产生一个输出流到1个或多个输出topic，有效地将输入流转换到输出流。 4.Connector API，允许构建运行可重复使用的生产者或消费者，将topic和现有的应用程序或数据系统连接起来。例如，一个关系型数据库的连接器可以捕获到该库下每一个表的变化。  ##二、Producer发送数据到kafka的流程 DRS同步到kafka：DRS作为kafka的客户端，利用kafka的Producer API；将源端数据库产生的增量数据写入到目标kafka的topic上。 ##### 以mysql到kafka为例，大致流程入下： 1.将源端mysql的binlog日志记录的增量数据作为消息封装成一个Record。 2.经过拦截器，对消息进行过滤。 3.经过序列化器，将消息的key和value进行序列化，当然可以自行定义序列化规则或者自行编写序列化器。 4.消息经过分区器，确定这条消息需要发送到目标topic的分区号。如果在消息里面指定了partion字段，那么就是将消息发送到指定分区。 5.之后消息会封装从成一个一个批次汇总到RecordAccumulator。accumulator可以作为一个缓存，是kafka强大的写入性能原因之一。 6.之后会依赖一个后台唤醒的Sender线程，将数据有序的发送到leader partition所在的broker(kafka集群的每一个服务器都是一个broker)中。 7.在发送消息的过程中，kafka客户端可以从任意一个broker获取到kafka集群的metadata信息，metadata信息里面记录了kafka集群的每个topic的所有partition的信息: leader, fellow, isr, replicas等。 整体的流程如下图所示  ## 三、Producer两个重要参数 1.acks决定了生产者如何在性能与数据可靠之间做取舍，官方源码中描述如下： ``` public static final String ACKS_CONFIG = "acks"; private static final String ACKS_DOC = "The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the " + " durability of records that are sent. The following settings are allowed: " + "" + "acks=0 If set to zero then the producer will not wait for any acknowledgment from the" + " server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be" + " made that the server has received the record in this case, and the retries configuration will not" + " take effect (as the client won't generally know of any failures). The offset given back for each record will" + " always be set to -1." + "acks=1 This will mean the leader will write the record to its local log but will respond" + " without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after" + " acknowledging the record but before the followers have replicated it then the record will be lost." + "acks=all This means the leader will wait for the full set of in-sync replicas to" + " acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica" + " remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting." + ""; ``` 对于kafka来说，消息日志是按照topic分类存储的，而对于一个topic来说有partitons分区数，replication-factor副本数。 对于一个topic而言有多个分区，一个分又可以有多个副本。这些副本中，只有一个leader partition。其他都是follower partiton，仅有leader partition可以对外提供服务，follower partiton主要用于冗余备份。 而副本是存放在不同的broker上面的，因此在创建topic的时候，副本数不能大于broker的节点数的。 而acks参数呢，就是和副本有关系。 ``` acks=0：这意味着producer发送数据后，不会等待broker确认，直接发送下一条数据，性能最好 acks=1：为1意味着producer发送数据后，需要等待leader副本确认接收后，才会发送下一条数据，性能次之 acks=-1/all：这个代表的是all，意味着发送的消息写入leader partition后，等到follower从leader拉取到消息后，才会发送下一条数据，性能最差，但可靠性最强 ``` 而DRS以可靠性优先，因此设置的acks参数值为all，确保消息写入到所有可用副本后，才进行下一条写入。 2.max.in.flight.requests.per.connection，官方源码描述如下： ``` public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection"; private static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC = "The maximum number of unacknowledged requests the client will send on a single connection before blocking." + " Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of" + " message re-ordering due to retries (i.e., if retries are enabled)."; ``` ``` // 在InFlightRequests.java中 /** * Can we send more requests to this node? * * @param node Node in question * @return true iff we have no requests still being sent to the given node */ public boolean canSendMore(String node) { Deque queue = requests.get(node); return queue == null || queue.isEmpty() || (queue.peekFirst().send.completed() && queue.size() this.maxInFlightRequestsPerConnection); } ``` ``` // 在Sender.java中 if (guaranteeMessageOrder) { // Mute all the partitions drained for (List batchList : batches.values()) { for (ProducerBatch batch : batchList) this.accumulator.mutePartition(batch.topicPartition); } } ``` max.in.flight.requests.per.connection表示在单个连接中，最多可以忍受多少个请求处于发送中没有没有响应。kafka源码中这个参数默认是5，可以认为，在一个连接中有5个请求发送出去了，并且Producer都没有收到broker的响应。 如果这个参数大于1，由于有重试机制，可能会存在消息顺序错乱的风险。 如下图，在一个网络连接中将batch封装成不同的request，从batch队列中取出数据，按照顺序封装成不同的request(请求1... 请求5).  如果broker在处理请求2时因为borker节点不可以等因素导致写消息到partition异常了，但是其它请求的数据都正常写入了。此时由于重试机制，Producer会将请求2重新发送。 导致broker写入到leader partition消息顺序错乱。  而DRS为了保证数据写入到kafka是有序的，max.in.flight.requests.per.connection参数设置为1，但是这样降低了kafka的吞吐量。

【功能模块】mrs-flink【操作步骤&问题现象】1、使用per-job模式上传样例代码到集群上运行，执行FemaleInfoCollectionFromKafka这个样例；2、flink任务创建成功，但是任务执行时报错，显示KafkaConsumer的一个方法不存在。【截图信息】报错信息：客户端lib里已补充相关jar包，不知道具体缺少哪个？【日志信息】（可选，上传日志内容或者附件）org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)    at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)    at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)    at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)    at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:484)    at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:380)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:279)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:194)    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at akka.actor.Actor$class.aroundReceive(Actor.scala:517)    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)    at akka.actor.ActorCell.invoke(ActorCell.scala:561)    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)    at akka.dispatch.Mailbox.run(Mailbox.scala:225)    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)    at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)    at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)    at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)    at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)Caused by: java.lang.NoSuchMethodError: org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.(Lorg/slf4j/Logger;Lorg/apache/flink/streaming/connectors/kafka/internal/Handover;Ljava/util/Properties;Lorg/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue;Ljava/lang/String;JZLorg/apache/flink/metrics/MetricGroup;Lorg/apache/flink/metrics/MetricGroup;)V    at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.(KafkaFetcher.java:109)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.createFetcher(FlinkKafkaConsumer.java:237)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:695)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:196)

一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询：   cat /etc/os-release 2.软件版本   Kafka版本为：Kafka 2.2.0 3.JDK依赖   Kafka 是用Scala 语言开发的，运行在JVM上，因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境，需要自己自行安装。   这里安装JDK环境做简单说明，请开发者自行安装，我这里安装的是 openjdk version "1.8.0_242"。   yum install java    输入 y   查看JDK版本   java -version二、详细安装步骤1.在根目录创建data目录   mkdir data2.进入data目录并下载Kafka 2.2.0  cd data   wget https://archive.apache.org/dist/kafka/2.2.0/kafka_2.12-2.2.0.tgz3.解压并进入到解压后的目录tar -zvxf kafka_2.12-2.2.0.tgz   cd kafka_2.12-2.2.04.启动zookeeper服务和kafka服务，可以在命令的结尾加个&符号，这样服务就可以在后台运行   ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &   ./bin/kafka-server-start.sh ./config/server.properties &     三、验证1.创建一个叫"euler"的topic，它只有一个分区，一个副本：   ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic euler2.发送消息，运行producer并在控制台中输一些消息，这些消息将被发送到服务端   ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic euler3.在另一个终端开启consumer，可以读取到刚才发出的消息并输出   ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic euler 至此所有操作已完成，请指正。