java B2B2C源码电子商城系统

  • 时间:
  • 浏览:1

将会大伙 在启动Kafka时均采用了默认配置,越来越大伙 不都要再做任何过后 配置就能在本地实现从RabbitMQ到Kafka的切换。大伙 都都要尝试把前一天搭建的ZooKeeper、Kafka启动起来,并将修改为spring-cloud-starter-bus-kafka模块的config-server和config-client启动起来。

大伙 对Kafka有了过后 基本了解前一天,下面大伙 来尝试构建另另有另一个Kafka服务端,并体验一下基于Kafka的消息生产与消费。

环境安装

首先,大伙 都要从官网上下载安装介质。都要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码 壹零叁八柒柒四六二六

在解压Kafka的安装包前一天,都都要看后其目录社会形态如下:

在启动了config-server和config-client前一天,为了更明显地观察消息总线刷新配置的效果,大伙 都都要在本地启动多个不同端口的config-client。此时,大伙 的config-server以及多个config-client都将会连接到了由Kafka实现的消息总线上。大伙 都都要先访问各个config-client上的/from请求,查看他获取到的配置内容。可是,修改Git中对应的参数内容,再访问各个config-client上的/from请求,都都要看后配置内容并越来越改变。最后,大伙 向config-server发送POST请求:/bus/refresh,此时大伙 再去访问各个config-client上的/from请求,就能获得到最新的配置信息,各客户端上的配置都将会加载为最新的Git配置内容。

从控制台信息中,大伙 都都要看后ZooKeeper从指定的config/zookeeper.properties配置文件中读取信息并绑定2181端口启动服务。有前一天启动失败,可查看一下端口算不算被占用,都都要杀掉占用程序运行或通过修改config/zookeeper.properties配置文件中的clientPort内容以绑定过后 端口号来启动ZooKeeper。

此时,大伙 都都要使用kafka-topics --list --zookeeper localhost:2181命令来查看当前Kafka中的Topic,若已成功启动了config-server并配置正确,大伙 就都都要在Kafka中看后将会多了另另有另一个名为springCloudBus的Topic。

启动Kafka,执行命令:kafka-server-start config/server.properties,该命令也都要指定Kafka配置文件的正确位置,如上命令中指向了解压目录中有 的默认配置。若在测试时,使用内部管理集中环境的ZooKeeper一段话,大伙 都都要在该配置文件中通过zookeeper.connect参数来设置ZooKeeper的地址和端口,它默认会连接本地2181端口的ZooKeeper;将会都要设置多个ZooKeeper节点,都都要为过后 参数配置多个ZooKeeper地址,并用逗号分割。比如:zookeeper.connect=127.0.0.1:5000,127.0.0.1:5001,127.0.0.1:5002。

启动ZooKeeper,执行命令:zookeeper-server-start config/zookeeper.properties,该命令都要指定zookeeper的配置文件位置都可不里能 正确启动,kafka的压缩包中中有 了其默认配置,开发与测试环境不都要修改。

大伙 再启动配置了spring-cloud-starter-bus-kafka模块的config-client,都都要看后控制台中输出如下内容:

另外,将会大伙 不使用kafka-topics命令来手工创建,直接进行下面的内容进行消息创建时也会自动创建Topics来使用。

都都要看后,config-client启动时输出了相似的内容,大伙 都订阅了名为springCloudBus的Topic。

启动测试

下面大伙 来尝试启动ZooKeeper和Kafka来进行消息的生产和消费。示例中所有的命令均已配置了Kafka的环境变量为例。

在config-server启动时,大伙 都都要在控制台中看后如下输出:

将会Kafka的设计中依赖了ZooKeeper,可是 大伙 都都要在bin和config目录中除了看后Kafka相关的内容之外,还有ZooKeeper相关的内容。其中bin目录存放了Kafka和ZooKeeper的命令行工具,bin根目录下是适用于Linux/Unix的shell,而bin/windows下的则是适用于windows下的bat。大伙 都都要根据实际的系统来设置环境变量,以方便后续的使用和操作。而在config目录中,则是用来存放了关于Kafka与ZooKeeper的配置信息。

创建消息消费者,执行命令:kafka-console-consumer --zookeeper localhost:2181 --topic test --from-beginning。kafka-console-consumer命令启动的是Kafka基于命令行的消息消费客户端,在启动前一天,大伙 马上都都要在控制台中看后输出了前一天大伙 在消息生产客户端中发送的消息。大伙 都都要再次打开前一天的消息生产客户端来发送消息,并观察消费者这边对消息的输出来体验Kafka对消息的基础处里。

RefreshListener监听类记录了收到远程刷新请求,并刷新了from属性的日志。

在上边的例子中,将会Kafka、ZooKeeper均运行于本地,可是 大伙 越来越在测试程序运行中通过配置信息来指定Kafka和ZooKeeper的配置信息,就完成了本地消息总线的试验。可是大伙 实际应用中,Kafka和ZooKeeper一般都会独立部署,可是 在应用中都都要来为Kafka和ZooKeeper配置过后 连接信息等。Kafka的整合与RabbitMQ不同,在Spring Boot 1.3.7中并越来越直接提供的Starter模块,可是采用了Spring Cloud Stream的Kafka模块,可是 对于Kafka的配置均采用了spring.cloud.stream.kafka的前缀。java B2B2C springmvc mybatis多租户电子商城系统

从控制台的输出内容,大伙 都都要看后config-server连接到了Kafka中,并使用了名为springCloudBus的Topic。

创建消息生产者,执行命令:kafka-console-producer --broker-list localhost:9092 --topic test。kafka-console-producer命令都都要启动Kafka基于命令行的消息生产客户端,启动后都都要直接在控制台中输入消息来发送,控制台中的每一行数据都会被视为一根绳子 消息来发送。大伙 都都要尝试输入几行消息,将会此时并越来越消费者,可是 哪此输入的消息都会被阻塞在名为test的Topics中,直到有消费者将其消费掉位置。

创建Topic,执行命令:kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test,通过该命令,创建另另有另一个名为“test”的Topic,该Topic中有 另另有另一个分区另另有另一个Replica。在创建完成后,都都要使用kafka-topics --list --zookeeper localhost:2181命令来查看当前的Topic。

从config-client的控制台中,大伙 都都要看后如下内容:

整合Spring Cloud Bus

在上一篇使用Rabbit实现消息总线的案例中,大伙 将会通过引入spring-cloud-starter-bus-amqp模块,完成了使用RabbitMQ来实现的消息总线。若大伙 要使用Kafka来实现消息总线时,只都要把spring-cloud-starter-bus-amqp替加进spring-cloud-starter-bus-kafka模块,在pom.xml的dependenies节点中进行修改,具体如下: