定制软件Kafka安装与配置

1.下载Kafka2.13-3.1.0

最新版为 

下载Zookeper

最新版为 

2.单机安装zookeper

依赖于zookeeper,定制软件官方承诺将来会移除.

解压文件:

  1. tar zxvf apache-zookeeper-3.8.0-bin.tar.gz -C /opt/
  2. mv /opt/apache-zookeeper-3.8.0-bin/ /opt/zookeeper

在/opt// 定制软件目录下创建数据文件目定制软件录和日志文件目录

  1. mkdir /opt/zookeeper/zkData
  2. mkdir /opt/zookeeper/zkLog

# 定制软件复制一份配置文件并修改

  1. cd /opt/zookeeper/conf/
  2. cp zoo_sample.cfg zoo.cfg
  3. vi zoo.cfg
  1. # 定制软件修改如下内容
  2. dataDir=/opt/zookeeper/zkData
  3. dataLogDir=/opt/zookeeper/zkLog

启动

  1. cd /opt/zookeeper/bin/
  2. # 启动zookeeper
  3. ./zkServer.sh start
  4. # 定制软件查看进程是否启动
  5. jps
  6. # 查看状态
  7. ./zkServer.sh status
  8. # 停止zookeeper
  9. ./zkServer.sh stop

3.安装Kafka

定制软件解压到指定目录

  1. cd /home
  2. $ tar -xzf kafka_2.13-3.1.0.tgz
  3. $ cd kafka_2.13-3.1.0

修改config目录下vi server.propertie文件

  1. listeners = PLAINTEXT://192.168.2.40:9092
  2.  
  3. #定制软件多个可用逗号分隔
  4.  
  5. #zookeeper.connect=server1:2181,server2:2181,server3:2181
  6.  
  7. zookeeper.connect=192.168.2.40:2181

启动命令:

bin/kafka-server-start.sh config/server.properties

定制软件此方式可以实时查看日志.

定制软件后台启动方式:

./kafka-server-start.sh -daemon ../config/server.properties

定制软件查询进程和关闭命令

  1. jps
  2. ./kafka-server-stop.sh

登录zookeeper客户端，查看/brokers/ids

  1. cd /opt/zookeeper/bin/
  2. zkCli.sh
  3. # 查询结果如下：
  4. [zk: localhost:2181(CONNECTED) 0] ls /brokers/ids
  5. [0]
  6. [zk: localhost:2181(CONNECTED) 1] quit

kafka常见命令

  1. #创建主题 主题名是 quickstart-events
  2. $ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
  3. #查询主题
  4. $ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
  5. #主题中写入消息
  6.  bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
  7. This is my first event
  8. This is my second event
  9. #主题中读取消息
  10.  bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
  11. This is my first event
  12. This is my second event

kafka集群

假如现在有两台服务器192.168.2.40,192.168.2.41

kafka的安装与配置如上,两台服务器唯一不同的地方就是配置文件中的broker.id和listeners的配置

 修改config目录下vi server.propertie文件

192.168.2.40

  1. listeners = PLAINTEXT://192.168.2.40:9092
  2. broker.id=0

 

192.168.2.41

  1. listeners = PLAINTEXT://192.168.2.41:9092
  2. broker.id=1

bin目录启动命令都添加

  1. vi kafka-server-start.sh
  2. #添加  export JMX_PORT="9999"
  3. if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
  4.     export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
  5.     export JMX_PORT="9999"
  6. fi

 登录zookeeper客户端，查看/brokers/ids

4.可视化工具kafka-eagle

下载:

解压

  1. cd /home
  2. tar -zxvf efak-web-2.1.0-bin.tar.gz

在/etc/profile文件中添加环境变量KE_HOME 

  1. vi /etc/profile
  2. # 在profile文件中添加
  3. #解压后的efak目录
  4. export KE_HOME=/home/efak-web-2.1.0
  5. export PATH=$PATH:$KE_HOME/bin
  6. # 使修改后的profile文件生效
  7. . /etc/profile

安装MySQL并添加数据库ke，kafka-eagle之后会用到它；
修改配置文件$KE_HOME/conf/system-config.properties，主要是修改Zookeeper的配置和数据库配置，注释掉sqlite配置，改为使用MySQL.

  1. ######################################
  2. kafka.eagle.zk.cluster.alias=cluster1
  3. cluster1.zk.list=localhost:2181
  4.  ######################################
  5. # kafka eagle webui port
  6. ######################################
  7. kafka.eagle.webui.port=8048
  8.  ######################################
  9. # kafka mysql jdbc driver address
  10. ######################################
  11. kafka.eagle.driver=com.mysql.cj.jdbc.Driver
  12. kafka.eagle.url=jdbc:mysql://localhost:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
  13. kafka.eagle.username=root
  14. kafka.eagle.password=123456

 启动后会自动使用上面的数据库连接，创建并初始化数据库.名称ke.有时候ke数据库会初始化失败，这时可以手动去执行创建数据库的脚本。脚本如下:

  1.  
  2.  
  3. /*!40101 SET NAMES utf8 */;
  4.  
  5. /*!40101 SET SQL_MODE=''*/;
  6.  
  7. /*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
  8. /*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
  9. /*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
  10. /*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;
  11. CREATE DATABASE /*!32312 IF NOT EXISTS*/`ke` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */;
  12.  
  13. USE `ke`;
  14.  
  15. /*Table structure for table `ke_alarm_clusters` */
  16.  
  17. DROP TABLE IF EXISTS `ke_alarm_clusters`;
  18.  
  19. CREATE TABLE `ke_alarm_clusters` (
  20.   `id` bigint NOT NULL AUTO_INCREMENT,
  21.   `type` varchar(32) DEFAULT NULL,
  22.   `cluster` varchar(64) DEFAULT NULL,
  23.   `server` text,
  24.   `alarm_group` varchar(128) DEFAULT NULL,
  25.   `alarm_times` int DEFAULT NULL,
  26.   `alarm_max_times` int DEFAULT NULL,
  27.   `alarm_level` varchar(4) DEFAULT NULL,
  28.   `is_normal` varchar(2) DEFAULT 'Y',
  29.   `is_enable` varchar(2) DEFAULT 'Y',
  30.   `created` varchar(32) DEFAULT NULL,
  31.   `modify` varchar(32) DEFAULT NULL,
  32.   PRIMARY KEY (`id`)
  33. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  34.  
  35. /*Data for the table `ke_alarm_clusters` */
  36.  
  37. /*Table structure for table `ke_alarm_config` */
  38.  
  39. DROP TABLE IF EXISTS `ke_alarm_config`;
  40.  
  41. CREATE TABLE `ke_alarm_config` (
  42.   `cluster` varchar(64) NOT NULL,
  43.   `alarm_group` varchar(128) NOT NULL,
  44.   `alarm_type` varchar(16) DEFAULT NULL,
  45.   `alarm_url` text,
  46.   `http_method` varchar(16) DEFAULT NULL,
  47.   `alarm_address` text,
  48.   `created` varchar(32) DEFAULT NULL,
  49.   `modify` varchar(32) DEFAULT NULL,
  50.   PRIMARY KEY (`cluster`,`alarm_group`)
  51. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  52.  
  53. /*Data for the table `ke_alarm_config` */
  54.  
  55. /*Table structure for table `ke_alarm_consumer` */
  56.  
  57. DROP TABLE IF EXISTS `ke_alarm_consumer`;
  58.  
  59. CREATE TABLE `ke_alarm_consumer` (
  60.   `id` bigint NOT NULL AUTO_INCREMENT,
  61.   `cluster` varchar(64) DEFAULT NULL,
  62.   `group` varchar(128) DEFAULT NULL,
  63.   `topic` varchar(128) DEFAULT NULL,
  64.   `lag` bigint DEFAULT NULL,
  65.   `alarm_group` varchar(128) DEFAULT NULL,
  66.   `alarm_times` int DEFAULT NULL,
  67.   `alarm_max_times` int DEFAULT NULL,
  68.   `alarm_level` varchar(4) DEFAULT NULL,
  69.   `is_normal` varchar(2) DEFAULT 'Y',
  70.   `is_enable` varchar(2) DEFAULT 'Y',
  71.   `created` varchar(32) DEFAULT NULL,
  72.   `modify` varchar(32) DEFAULT NULL,
  73.   PRIMARY KEY (`id`)
  74. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  75.  
  76. /*Data for the table `ke_alarm_consumer` */
  77.  
  78. /*Table structure for table `ke_alarm_crontab` */
  79.  
  80. DROP TABLE IF EXISTS `ke_alarm_crontab`;
  81.  
  82. CREATE TABLE `ke_alarm_crontab` (
  83.   `id` bigint NOT NULL,
  84.   `type` varchar(64) NOT NULL,
  85.   `crontab` varchar(32) DEFAULT NULL,
  86.   `is_enable` varchar(2) DEFAULT 'Y',
  87.   `created` varchar(32) DEFAULT NULL,
  88.   `modify` varchar(32) DEFAULT NULL,
  89.   PRIMARY KEY (`id`,`type`)
  90. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  91.  
  92. /*Data for the table `ke_alarm_crontab` */
  93.  
  94. /*Table structure for table `ke_connect_config` */
  95.  
  96. DROP TABLE IF EXISTS `ke_connect_config`;
  97.  
  98. CREATE TABLE `ke_connect_config` (
  99.   `id` bigint NOT NULL AUTO_INCREMENT,
  100.   `cluster` varchar(64) DEFAULT NULL,
  101.   `connect_uri` varchar(128) DEFAULT NULL,
  102.   `version` varchar(32) DEFAULT NULL,
  103.   `alive` varchar(16) DEFAULT NULL,
  104.   `created` varchar(32) DEFAULT NULL,
  105.   `modify` varchar(32) DEFAULT NULL,
  106.   PRIMARY KEY (`id`)
  107. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  108.  
  109. /*Data for the table `ke_connect_config` */
  110.  
  111. /*Table structure for table `ke_consumer_bscreen` */
  112.  
  113. DROP TABLE IF EXISTS `ke_consumer_bscreen`;
  114.  
  115. CREATE TABLE `ke_consumer_bscreen` (
  116.   `cluster` varchar(64) DEFAULT NULL,
  117.   `group` varchar(128) DEFAULT NULL,
  118.   `topic` varchar(64) DEFAULT NULL,
  119.   `logsize` bigint DEFAULT NULL,
  120.   `difflogsize` bigint DEFAULT NULL,
  121.   `offsets` bigint DEFAULT NULL,
  122.   `diffoffsets` bigint DEFAULT NULL,
  123.   `lag` bigint DEFAULT NULL,
  124.   `timespan` bigint DEFAULT NULL,
  125.   `tm` varchar(16) DEFAULT NULL,
  126.   KEY `idx_timespan` (`timespan`),
  127.   KEY `idx_tm_cluster_diffoffsets` (`tm`,`cluster`,`diffoffsets`)
  128. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  129.  
  130. /*Data for the table `ke_consumer_bscreen` */
  131.  
  132. /*Table structure for table `ke_consumer_group` */
  133.  
  134. DROP TABLE IF EXISTS `ke_consumer_group`;
  135.  
  136. CREATE TABLE `ke_consumer_group` (
  137.   `cluster` varchar(64) NOT NULL,
  138.   `group` varchar(128) NOT NULL,
  139.   `topic` varchar(128) NOT NULL,
  140.   `status` int DEFAULT NULL,
  141.   PRIMARY KEY (`cluster`,`group`,`topic`)
  142. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  143.  
  144. /*Data for the table `ke_consumer_group` */
  145.  
  146. /*Table structure for table `ke_consumer_group_summary` */
  147.  
  148. DROP TABLE IF EXISTS `ke_consumer_group_summary`;
  149.  
  150. CREATE TABLE `ke_consumer_group_summary` (
  151.   `cluster` varchar(64) NOT NULL,
  152.   `group` varchar(128) NOT NULL,
  153.   `topic_number` varchar(128) NOT NULL,
  154.   `coordinator` varchar(128) DEFAULT NULL,
  155.   `active_topic` int DEFAULT NULL,
  156.   `active_thread_total` int DEFAULT NULL,
  157.   PRIMARY KEY (`cluster`,`group`)
  158. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  159.  
  160. /*Data for the table `ke_consumer_group_summary` */
  161.  
  162. /*Table structure for table `ke_logsize` */
  163.  
  164. DROP TABLE IF EXISTS `ke_logsize`;
  165.  
  166. CREATE TABLE `ke_logsize` (
  167.   `cluster` varchar(64) DEFAULT NULL,
  168.   `topic` varchar(64) DEFAULT NULL,
  169.   `logsize` bigint DEFAULT NULL,
  170.   `diffval` bigint DEFAULT NULL,
  171.   `timespan` bigint DEFAULT NULL,
  172.   `tm` varchar(16) DEFAULT NULL,
  173.   KEY `idx_timespan` (`timespan`),
  174.   KEY `idx_tm_topic` (`tm`,`topic`),
  175.   KEY `idx_tm_cluster_diffval` (`tm`,`cluster`,`diffval`)
  176. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  177.  
  178. /*Data for the table `ke_logsize` */
  179.  
  180. /*Table structure for table `ke_metrics` */
  181.  
  182. DROP TABLE IF EXISTS `ke_metrics`;
  183.  
  184. CREATE TABLE `ke_metrics` (
  185.   `cluster` varchar(64) DEFAULT NULL,
  186.   `broker` text,
  187.   `type` varchar(32) DEFAULT NULL,
  188.   `key` varchar(64) DEFAULT NULL,
  189.   `value` varchar(128) DEFAULT NULL,
  190.   `timespan` bigint DEFAULT NULL,
  191.   `tm` varchar(16) DEFAULT NULL
  192. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  193.  
  194. /*Data for the table `ke_metrics` */
  195.  
  196. /*Table structure for table `ke_metrics_offline` */
  197.  
  198. DROP TABLE IF EXISTS `ke_metrics_offline`;
  199.  
  200. CREATE TABLE `ke_metrics_offline` (
  201.   `cluster` varchar(64) NOT NULL,
  202.   `key` varchar(128) NOT NULL,
  203.   `one` varchar(128) DEFAULT NULL,
  204.   `mean` varchar(128) DEFAULT NULL,
  205.   `five` varchar(128) DEFAULT NULL,
  206.   `fifteen` varchar(128) DEFAULT NULL,
  207.   PRIMARY KEY (`cluster`,`key`)
  208. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  209.  
  210. /*Data for the table `ke_metrics_offline` */
  211.  
  212. /*Table structure for table `ke_p_role` */
  213.  
  214. DROP TABLE IF EXISTS `ke_p_role`;
  215.  
  216. CREATE TABLE `ke_p_role` (
  217.   `id` bigint NOT NULL AUTO_INCREMENT,
  218.   `name` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'role name',
  219.   `seq` tinyint NOT NULL COMMENT 'rank',
  220.   `description` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'role describe',
  221.   PRIMARY KEY (`id`)
  222. ) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  223.  
  224. /*Data for the table `ke_p_role` */
  225.  
  226. insert  into `ke_p_role`(`id`,`name`,`seq`,`description`) values 
  227. (1,'Administrator',1,'Have all permissions'),
  228. (2,'Devs',2,'Own add or delete'),
  229. (3,'Tourist',3,'Only viewer');
  230.  
  231. /*Table structure for table `ke_resources` */
  232.  
  233. DROP TABLE IF EXISTS `ke_resources`;
  234.  
  235. CREATE TABLE `ke_resources` (
  236.   `resource_id` bigint NOT NULL AUTO_INCREMENT,
  237.   `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'resource name',
  238.   `url` varchar(255) NOT NULL,
  239.   `parent_id` int NOT NULL,
  240.   PRIMARY KEY (`resource_id`)
  241. ) ENGINE=InnoDB AUTO_INCREMENT=25 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  242.  
  243. /*Data for the table `ke_resources` */
  244.  
  245. insert  into `ke_resources`(`resource_id`,`name`,`url`,`parent_id`) values 
  246. (1,'System','/system',-1),
  247. (2,'User','/system/user',1),
  248. (3,'Role','/system/role',1),
  249. (4,'Resource','/system/resource',1),
  250. (5,'Notice','/system/notice',1),
  251. (6,'Topic','/topic',-1),
  252. (7,'Message','/topic/message',6),
  253. (8,'Create','/topic/create',6),
  254. (9,'Alarm','/alarm',-1),
  255. (10,'Add','/alarm/add',9),
  256. (11,'Modify','/alarm/modify',9),
  257. (12,'Cluster','/cluster',-1),
  258. (13,'ZkCli','/cluster/zkcli',12),
  259. (14,'UserDelete','/system/user/delete',1),
  260. (15,'UserModify','/system/user/modify',1),
  261. (16,'Mock','/topic/mock',6),
  262. (18,'Create','/alarm/create',9),
  263. (19,'History','/alarm/history',9),
  264. (20,'Manager','/topic/manager',6),
  265. (21,'PasswdReset','/system/user/reset',1),
  266. (22,'Config','/alarm/config',9),
  267. (23,'List','/alarm/list',9),
  268. (24,'Hub','/topic/hub',6);
  269.  
  270. /*Table structure for table `ke_role_resource` */
  271.  
  272. DROP TABLE IF EXISTS `ke_role_resource`;
  273.  
  274. CREATE TABLE `ke_role_resource` (
  275.   `id` bigint NOT NULL AUTO_INCREMENT,
  276.   `role_id` int NOT NULL,
  277.   `resource_id` int NOT NULL,
  278.   PRIMARY KEY (`id`)
  279. ) ENGINE=InnoDB AUTO_INCREMENT=26 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  280.  
  281. /*Data for the table `ke_role_resource` */
  282.  
  283. insert  into `ke_role_resource`(`id`,`role_id`,`resource_id`) values 
  284. (1,1,1),
  285. (2,1,2),
  286. (3,1,3),
  287. (4,1,4),
  288. (5,1,5),
  289. (6,1,7),
  290. (7,1,8),
  291. (8,1,10),
  292. (9,1,11),
  293. (10,1,13),
  294. (11,2,7),
  295. (12,2,8),
  296. (13,2,13),
  297. (14,2,10),
  298. (15,2,11),
  299. (16,1,14),
  300. (17,1,15),
  301. (18,1,16),
  302. (19,1,18),
  303. (20,1,19),
  304. (21,1,20),
  305. (22,1,21),
  306. (23,1,22),
  307. (24,1,23),
  308. (25,1,24);
  309.  
  310. /*Table structure for table `ke_sql_history` */
  311.  
  312. DROP TABLE IF EXISTS `ke_sql_history`;
  313.  
  314. CREATE TABLE `ke_sql_history` (
  315.   `id` bigint NOT NULL AUTO_INCREMENT,
  316.   `cluster` varchar(64) DEFAULT NULL,
  317.   `username` varchar(64) DEFAULT NULL,
  318.   `host` varchar(128) DEFAULT NULL,
  319.   `ksql` text,
  320.   `status` varchar(16) DEFAULT NULL,
  321.   `spend_time` bigint DEFAULT NULL,
  322.   `created` varchar(32) DEFAULT NULL,
  323.   `tm` varchar(16) DEFAULT NULL,
  324.   PRIMARY KEY (`id`)
  325. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  326.  
  327. /*Data for the table `ke_sql_history` */
  328.  
  329. /*Table structure for table `ke_topic_rank` */
  330.  
  331. DROP TABLE IF EXISTS `ke_topic_rank`;
  332.  
  333. CREATE TABLE `ke_topic_rank` (
  334.   `cluster` varchar(64) NOT NULL,
  335.   `topic` varchar(64) NOT NULL,
  336.   `tkey` varchar(64) NOT NULL,
  337.   `tvalue` bigint DEFAULT NULL,
  338.   PRIMARY KEY (`cluster`,`topic`,`tkey`)
  339. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  340.  
  341. /*Data for the table `ke_topic_rank` */
  342.  
  343. /*Table structure for table `ke_user_role` */
  344.  
  345. DROP TABLE IF EXISTS `ke_user_role`;
  346.  
  347. CREATE TABLE `ke_user_role` (
  348.   `id` bigint NOT NULL AUTO_INCREMENT,
  349.   `user_id` int NOT NULL,
  350.   `role_id` tinyint NOT NULL,
  351.   PRIMARY KEY (`id`)
  352. ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  353.  
  354. /*Data for the table `ke_user_role` */
  355.  
  356. insert  into `ke_user_role`(`id`,`user_id`,`role_id`) values 
  357. (1,1,1);
  358.  
  359. /*Table structure for table `ke_users` */
  360.  
  361. DROP TABLE IF EXISTS `ke_users`;
  362.  
  363. CREATE TABLE `ke_users` (
  364.   `id` bigint NOT NULL AUTO_INCREMENT,
  365.   `rtxno` int NOT NULL,
  366.   `username` varchar(64) NOT NULL,
  367.   `password` varchar(128) NOT NULL,
  368.   `email` varchar(64) NOT NULL,
  369.   `realname` varchar(128) NOT NULL,
  370.   PRIMARY KEY (`id`)
  371. ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
  372.  
  373. /*Data for the table `ke_users` */
  374.  
  375. insert  into `ke_users`(`id`,`rtxno`,`username`,`password`,`email`,`realname`) values 
  376. (1,1000,'admin','123456','admin@email.com','Administrator');
  377.  
  378. /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
  379. /*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
  380. /*!40014 SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS */;
  381. /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
  1. # 停止服务
  2. bin/ke.sh stop
  3. # 重启服务
  4. bin/ke.sh restart
  5. # 查看服务运行状态
  6. bin/ke.sh status
  7. # 查看服务状态
  8. bin/ke.sh stats
  9. # 动态查看服务输出日志
  10. tail -f /logs/ke_console.out 
  11. #查看进程号及端口号
  12. netstat -ntlp
  13. #通过PID查询出进程位置
  14. ps aux|grep 进程号
  15. #确定进程所在的目录
  16. ll /proc/进程号;

 ​脚本创建成功，再重新启动服务即可.
启动成功可以直接访问，输入账号密码admin:123456，访问地址：http://192.168.2.40:8048/

 

注意观察 brokers,topics的数量。brokers为0的话没有连接成功.

可视化工具自然少不了监控，如果你想开启kafka-eagle对Kafka的监控功能的话，需要修改Kafka的启动脚本，暴露JMX的端口.

  1. vi kafka-server-start.sh
  2. #添加  export JMX_PORT="9999"
  3. if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
  4.     export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
  5.     export JMX_PORT="9999"
  6. fi

kafka集群图示:

 

5.SpringBoot整合Kafka.

在pom.xml中添加

<dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka</artifactId></dependency>

 在application.yml中spring节点下添加

  1. spring
  2.     kafka:
  3.     bootstrap-servers: 192.168.2.40:9092
  4.     producer: # 生产者配置
  5.       retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
  6.       batch-size: 16384 #16K
  7.       buffer-memory: 33554432 #32M
  8.       acks: 1
  9.       # 指定消息key和消息体的编解码方式
  10.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  11.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  12.     consumer:
  13.       group-id: zhTestGroup # 消费者组
  14.       enable-auto-commit: false # 关闭自动提交
  15.       auto-offset-reset: earliest # 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
  16.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  17.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  18.     listener:
  19.       # 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
  20.       # RECORD
  21.       # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
  22.       # BATCH
  23.       # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
  24.       # TIME
  25.       # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
  26.       # COUNT
  27.       # TIME |　COUNT　有一个条件满足时提交
  28.       # COUNT_TIME
  29.       # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
  30.       # MANUAL
  31.       # 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
  32.       # MANUAL_IMMEDIATE
  33.       ack-mode: manual_immediate

生产者:

  1. @RestController
  2. public class KafkaProducer {
  3.  
  4. 	@Autowired
  5. 	private KafkaTemplate kafkaTemplate;
  6. 	// 发送消息
  7. 	@GetMapping("/kafka/normal/{message}")
  8. 	public void sendMessage1(@PathVariable("message") String normalMessage) {
  9. 		kafkaTemplate.send("quickstart-events", normalMessage);
  10. 	}
  11.  
  12. }

 消费者:

  1. @Component
  2. public class KafkaConsumer {
  3. 	// 消费监听
  4. 	@KafkaListener(topics = {"quickstart-events"})
  5. 	public void onMessage1(ConsumerRecord<?, ?> record,Consumer consumer,Acknowledgment ack){
  6. 		// 消费的哪个topic、partition的消息,打印出消息内容
  7. 		System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
  8. 		//同步提交
  9. 		//consumer.commitSync();
  10. 		ack.acknowledge();
  11. 	}
  12. }

 可以用postman进行测试，观察结果.

网站建设定制开发 软件系统开发定制 定制软件开发 软件开发定制 定制app开发 app开发定制 app开发定制公司 电商商城定制开发 定制小程序开发 定制开发小程序 客户管理系统开发定制 定制网站 定制开发 crm开发定制 开发公司 小程序开发定制 定制软件 收款定制开发 企业网站定制开发 定制化开发 android系统定制开发 定制小程序开发费用 定制设计 专注app软件定制开发 软件开发定制定制 知名网站建设定制 软件定制开发供应商 应用系统定制开发 软件系统定制开发 企业管理系统定制开发 系统定制开发