SpringCloud使用Kafka消息总线、Kafka的安装与测试（CentOS 7）

妖狐艹你老母 2022-05-14 05:46 385阅读 0赞

# SpringCloud使用Kafka消息总线、Kafka的安装与测试（CentOS 7） #

主要内容：

*  **CentOS7静态地址配置**
 *  **Kafka安装与测试**
 *  **SpringCloud使用Kafka做消息总线**

--------------------

### CentOS7静态地址配置 ###

打开配置文件：

> \[root@localhost network-scripts\]\# vi /etc/sysconfig/network-scripts/ifcfg-ens33

修改内容如下：

> TYPE=Ethernet  
> PROXY\_METHOD=none  
> BROWSER\_ONLY=no  
> BOOTPROTO=static \#修改  
> DEFROUTE=yes  
> IPV4\_FAILURE\_FATAL=no  
> IPV6INIT=yes  
> IPV6\_AUTOCONF=yes  
> IPV6\_DEFROUTE=yes  
> IPV6\_FAILURE\_FATAL=no  
> IPV6\_ADDR\_GEN\_MODE=stable-privacy  
> NAME=ens33  
> UUID=bd4786ea-5587-45de-ad19-77786fe0d6f9  
> DEVICE=ens33  
> ONBOOT=yes \#修改  
> IPADDR0=192.168.129.130 \#增加  
> PREFIXO0=24 \#增加  
> GATEWAY0=192.168.129.2 \#增加（虚拟机中网关）  
> DNS1=8.8.8.8 \#增加  
> DNS2=8.8.4.4 \#增加

重启网络

> \[root@localhost network-scripts\]\# service network restart

查看IP地址

> \[root@localhost network-scripts\]\# ip addr

安装JDK

> \[root@localhost home\]\# mkdir jdk  
> \[root@localhost home\]\# ls  
> jdk  
> \[root@localhost home\]\# cd jdk  
> \[root@localhost jdk\]\# ls  
> jdk-8u161-linux-x64.tar.gz

解压文件

> \[root@localhost jdk\]\# tar -zxvf jdk-8u161-linux-x64.tar.gz  
> \[root@localhost jdk\]\# ls  
> jdk1.8.0\_161 jdk-8u161-linux-x64.tar.gz

配置环境变量

> \[root@localhost ~\]\# vi /etc/profile

文本末尾加入以下内容

export JAVA_HOME=/home/jdk/jdk1.8.0_161
    export CLASSPATH=.
    export PATH=$PATH:${JAVA_HOME}/bin

配置生效

> \[root@localhost ~\]\# source /etc/profile  
> \[root@localhost ~\]\# java -version  
> java version “1.8.0\_161”  
> Java(TM) SE Runtime Environment (build 1.8.0\_161-b12)  
> Java HotSpot(TM) 64-Bit Server VM (build 25.161-b12, mixed mode)

### Kafka安装与测试 ###

> \[root@localhost kafka\]\# tar -zxvf kafka\_2.11-2.0.0.tgz  
> \[root@localhost kafka\_2.11-2.0.0\]\# ls  
> bin config libs LICENSE NOTICE site-docs  
> \[root@localhost kafka\_2.11-2.0.0\]\# cd config  
> \[root@localhost config\]\# vi zookeeper.properties

可配置zookeeper的端口日志  
配置kafka:端口号，连接zookeeper等

> \[root@localhost kafka\_2.11-2.0.0\]\# cd config/  
> \[root@localhost config\]\# vi server-1.properties

启动zookeeper和kafka

> \[root@localhost kafka\_2.11-2.0.0\]\# nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties &  
> \[1\] 1119  
> \[root@localhost kafka\_2.11-2.0.0\]\# nohup: 忽略输入并把输出追加到”nohup.out”  
> \[root@localhost kafka\_2.11-2.0.0\]\# nohup ./bin/kafka-server-start.sh ./config/server-1.properties &  
> \[2\] 1408  
> \[root@localhost kafka\_2.11-2.0.0\]\# nohup: 忽略输入并把输出追加到”nohup.out”  
> \[root@localhost kafka\_2.11-2.0.0\]\# jps  
> 1408 Kafka  
> 1744 Jps  
> 1119 QuorumPeerMain

创建topic、查看topic状态

> \[root@localhost kafka\_2.11-2.0.0\]\# bin/kafka-topics.sh –create –zookeeper 192.168.129.130:2181 –replication-factor 1 –partitions 1 –topic test  
> Created topic “test”.  
> \[root@localhost kafka\_2.11-2.0.0\]\# bin/kafka-topics.sh -zookeeper 192.168.129.130:2181 -list  
> test  
> \[root@localhost kafka\_2.11-2.0.0\]\# bin/kafka-topics.sh –zookeeper 192.168.129.130:2181 –describe –topic test  
> Topic:test PartitionCount:1 ReplicationFactor:1 Configs:  
> Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0

发布消息

> \[root@localhost kafka\_2.11-2.0.0\]\# bin/kafka-console-producer.sh –broker-list 192.168.129.130:9092 –topic test

>oooo
    >aaaa

订阅消息

> \[root@localhost kafka\_2.11-2.0.0\]\# bin/kafka-console-consumer.sh –bootstrap-server 192.168.129.130:9092 –topic gxl –from-beginning  
> oooo  
> aaaa

之前遇到个报错：

> ERROR Error when sending message to topic test with key: null, value: 5 byt…  
> 原因：发布消息和订阅命令中kafka端口写错或者是要发布的topic不存在

### SpringCloud使用Kafka做消息总线 ###

总体架构：  
![这里写图片描述][70]  
1、注册中心地址：  
eureka.client.serviceUrl.defaultZone=[http://frog:10000/eureka][http_frog_10000_eureka]  
2、配置中心（configserver）：  
pom

<parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.5.6.RELEASE</version>
            <relativePath/> 
        </parent>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
            <spring-cloud.version>Dalston.RELEASE</spring-cloud.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-config-server</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-security</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-eureka</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-bus-kafka</artifactId>
            </dependency>
        </dependencies>
    
        
        <dependencyManagement>
            <dependencies>
                <dependency>
                    <groupId>org.springframework.cloud</groupId>
                    <artifactId>spring-cloud-dependencies</artifactId>
                    <version>${spring-cloud.version}</version>
                    <type>pom</type>
                    <scope>import</scope>
                </dependency>
            </dependencies>
        </dependencyManagement>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>

application.properties

spring.application.name=frog-config
    server.port=10010
    # 通过本地文件系统（指定位置）
    spring.profiles.active=native
    spring.cloud.config.server.native.search-locations=classpath:/properties/
    # 安全保护（添加security依赖）
    # 不使用随机密码（不配置默认使用随机密码），使用指定密码
    security.user.name=user
    security.user.password=1234567890
    # 服务化的配置中心
    eureka.client.serviceUrl.defaultZone=http://frog:10000/eureka
    # 服务注册中心实例的主机名
    eureka.instance.hostname=frog

kafka-dev.properties

#Kafka的服务端列表，默认localhost
    spring.cloud.stream.kafka.binder.brokers=192.168.129.130:9092
    #Kafka服务端的默认端口，当brokers属性中没有配置端口信息时，就会使用这个默认端口，默认9092
    spring.cloud.stream.kafka.binder.defaultBrokerPort=9092
    #Kafka服务端连接的ZooKeeper节点列表，默认localhost
    spring.cloud.stream.kafka.binder.zkNodes=192.168.129.130:2181
    #ZooKeeper节点的默认端口，当zkNodes属性中没有配置端口信息时，就会使用这个默认端口，默认2181
    spring.cloud.stream.kafka.binder.defaultZkPort=2181

frog-config-client-dev.properties

str=dev2.3333333333333

config-server项目结构：  
![这里写图片描述][70 1]  
配置中心需要启动后加载kafka-dev.properties配置文件中的配置信息需要在启动类中做如下改动：

@SpringBootApplication
    @EnableConfigServer
    @EnableDiscoveryClient
    public class ConfigApplication { 
    
        public static void main(String[] args) throws IOException {
            // 加载自定义配置文件
            Properties properties = new Properties();
            InputStream inputStream = ConfigApplication.class.getClassLoader().getResourceAsStream("properties/kafka-dev.properties");
            properties.load(inputStream);
            // 测试有没有加载
            System.out.println(properties.getProperty("spring.cloud.stream.kafka.binder.brokers"));
            SpringApplication sa = new SpringApplication(ConfigApplication.class);
            sa.setDefaultProperties(properties);
            sa.run(args);
        }
    }

启动  
![这里写图片描述][70 2]  
3、客户端（client）  
pom文件与configserver相同  
application.properties

server.port=10011
    
    # 配置中心设置账户密码，需要验证
    spring.cloud.config.username=user
    spring.cloud.config.password=1234567890
    
    spring.application.name=frog-config-client
    # 服务化的配置中心(启动时先启动注册中心，在启动配置中心server，稳定后启动服务端)
    eureka.client.serviceUrl.defaultZone=http://frog:10000/eureka
    # 服务注册中心实例的主机名
    eureka.instance.hostname=frog
    # 开启通过服务来访问config server的功能
    spring.cloud.config.discovery.enabled=true
    # 指定config server注册的服务名
    spring.cloud.config.discovery.service-id=frog-config
    # 指定配置中心的资源
    spring.cloud.config.profile=dev
    spring.cloud.config.label=master
    spring.cloud.config.name=frog-config-client,kafka
    
    # /refresh刷新时需要将验证关闭
    management.security.enabled=false

测试接口

@RestController
    @RefreshScope
    public class TestController { 
    
        @GetMapping("/test")
        public String test() {
            return this.str;
        }
    
        @Value("${str}")
        private String str;
    }

启动后请求/test看是否正常，因configserver配置了权限校验，需要输入用户名和密码  
![这里写图片描述][70 3]  
4、刷新配置  
修改frog-config-client-dev.properties配置文件中str值，再次请求client的/test发现没有刷新。  
通过postman请求configserver的/bus/refresh来刷新配置，头部信息添加权限校验信息。

![这里写图片描述][70 4]  
![这里写图片描述][70 5]  
刷新后再次请求客户端/test接口发现值发生改变。  
5、bug  
我目前使用的springcloud版本在请求配置中心/bus/refresh后会导致注册中心所有服务处于离线状态，使用Dalston新版本即可，比如：  
pom：

<spring-cloud.version>Dalston.SR5</spring-cloud.version>

[70]: /images/20220514/075598eb853542eab5827fa2dfbe8a34.png
[http_frog_10000_eureka]: http://frog:10000/eureka
[70 1]: /images/20220514/cf120fcce23a42449889b8f60b5fad1a.png
[70 2]: /images/20220514/7664ee4c170c4648b9bad473eddc3df6.png
[70 3]: /images/20220514/43c3d9ca317d485cb88946b70f5f252b.png
[70 4]: /images/20220514/8e48524be657405ca0135ca6111dac64.png
[70 5]: /images/20220514/3a94aec2c67c400587472c3b73ef4085.png