java B2B2C源码电子商城系统-Kafka快速入门-低调大师

java B2B2C源码电子商城系统-Kafka快速入门

2019-01-06 560

大家对Kafka有了一些基本了解之后，下面我们来尝试构建一个Kafka服务端，并体验一下基于Kafka的消息生产与消费。
环境安装
首先，我们需要从官网上下载安装介质。需要JAVA Spring Cloud大型企业分布式微服务云构建的B2B2C电子商务平台源码壹零叁八柒柒四六二六
在解压Kafka的安装包之后，可以看到其目录结构如下：

kafka
  +-bin
    +-windows
  +-config
  +-libs
  +-logs
  +-site-docs

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

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

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

[2016-09-28 08:05:34,849] INFO Reading configuration from: config\zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2016-09-28 08:05:34,850] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2016-09-28 08:05:34,851] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2016-09-28 08:05:34,851] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
...
[2016-09-28 08:05:34,940] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

从控制台信息中，我们可以看到ZooKeeper从指定的config/zookeeper.properties配置文件中读取信息并绑定2181端口启动服务。有时候启动失败，可查看一下端口是否被占用，可以杀掉占用进程或通过修改config/zookeeper.properties配置文件中的clientPort内容以绑定其他端口号来启动ZooKeeper。

启动Kafka，执行命令：kafka-server-start config/server.properties，该命令也需要指定Kafka配置文件的正确位置，如上命令中指向了解压目录包含的默认配置。若在测试时，使用外部集中环境的ZooKeeper的话，我们可以在该配置文件中通过zookeeper.connect参数来设置ZooKeeper的地址和端口，它默认会连接本地2181端口的ZooKeeper；如果需要设置多个ZooKeeper节点，可以为这个参数配置多个ZooKeeper地址，并用逗号分割。比如：zookeeper.connect=127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002。

创建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。

另外，如果我们不使用kafka-topics命令来手工创建，直接进行下面的内容进行消息创建时也会自动创建Topics来使用。

创建消息生产者，执行命令：kafka-console-producer --broker-list localhost:9092 --topic test。kafka-console-producer命令可以启动Kafka基于命令行的消息生产客户端，启动后可以直接在控制台中输入消息来发送，控制台中的每一行数据都会被视为一条消息来发送。我们可以尝试输入几行消息，由于此时并没有消费者，所以这些输入的消息都会被阻塞在名为test的Topics中，直到有消费者将其消费掉位置。

创建消息消费者，执行命令：kafka-console-consumer --zookeeper localhost:2181 --topic test --from-beginning。kafka-console-consumer命令启动的是Kafka基于命令行的消息消费客户端，在启动之后，我们马上可以在控制台中看到输出了之前我们在消息生产客户端中发送的消息。我们可以再次打开之前的消息生产客户端来发送消息，并观察消费者这边对消息的输出来体验Kafka对消息的基础处理。

整合Spring Cloud Bus
在上一篇使用Rabbit实现消息总线的案例中，我们已经通过引入spring-cloud-starter-bus-amqp模块，完成了使用RabbitMQ来实现的消息总线。若我们要使用Kafka来实现消息总线时，只需要把spring-cloud-starter-bus-amqp替换成spring-cloud-starter-bus-kafka模块，在pom.xml的dependenies节点中进行修改，具体如下：

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

如果我们在启动Kafka时均采用了默认配置，那么我们不需要再做任何其他配置就能在本地实现从RabbitMQ到Kafka的切换。我们可以尝试把刚刚搭建的ZooKeeper、Kafka启动起来，并将修改为spring-cloud-starter-bus-kafka模块的config-server和config-client启动起来。

在config-server启动时，我们可以在控制台中看到如下输出：

2016-09-28 22:11:29.627  INFO 15144 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder    : Using kafka topic for outbound: springCloudBus
2016-09-28 22:11:29.642  INFO 15144 --- [-localhost:2181] org.I0Itec.zkclient.ZkEventThread        : Starting ZkClient event thread.
...
016-09-28 22:11:30.290  INFO 15144 --- [           main] o.s.i.kafka.support.ProducerFactoryBean  : Using producer properties => {bootstrap.servers=localhost:9092, linger.ms=0, acks=1, compression.type=none, batch.size=16384}
2016-09-28 22:11:30.298  INFO 15144 --- [           main] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values:
...
2016-09-28 22:11:30.322  INFO 15144 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$1  : Adding {message-handler:outbound.springCloudBus} as a subscriber to the 'springCloudBusOutput' channel
2016-09-28 22:11:31.467  INFO 15144 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$7  : Adding {message-handler:inbound.springCloudBus.anonymous.8b9e6c7b-6a50-48c5-b981-8282a0d5a30b} as a subscriber to the 'bridge.springCloudBus' channel
2016-09-28 22:11:31.467  INFO 15144 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$7  : started inbound.springCloudBus.anonymous.8b9e6c7b-6a50-48c5-b981-8282a0d5a30b

从控制台的输出内容，我们可以看到config-server连接到了Kafka中，并使用了名为springCloudBus的Topic。

此时，我们可以使用kafka-topics --list --zookeeper localhost:2181命令来查看当前Kafka中的Topic，若已成功启动了config-server并配置正确，我们就可以在Kafka中看到已经多了一个名为springCloudBus的Topic。

我们再启动配置了spring-cloud-starter-bus-kafka模块的config-client，可以看到控制台中输出如下内容：

2016-09-28 22:43:55.067  INFO 6136 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder    : Using kafka topic for outbound: springCloudBus
2016-09-28 22:43:55.078  INFO 6136 --- [-localhost:2181] org.I0Itec.zkclient.ZkEventThread        : Starting ZkClient event thread.
...
2016-09-28 22:50:38.584  INFO 828 --- [           main] o.s.i.kafka.support.ProducerFactoryBean  : Using producer properties => {bootstrap.servers=localhost:9092, linger.ms=0, acks=1, compression.type=none, batch.size=16384}
2016-09-28 22:50:38.592  INFO 828 --- [           main] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
...
2016-09-28 22:50:38.615  INFO 828 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$1  : Adding {message-handler:outbound.springCloudBus} as a subscriber to the 'springCloudBusOutput' channel
2016-09-28 22:50:38.616  INFO 828 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'didispace:7002.springCloudBusOutput' has 1 subscriber(s).
2016-09-28 22:50:38.616  INFO 828 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$1  : started outbound.springCloudBus
...
2016-09-28 22:50:39.162  INFO 828 --- [           main] s.i.k.i.KafkaMessageDrivenChannelAdapter : started org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter@60cf855e
2016-09-28 22:50:39.162  INFO 828 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$7  : Adding {message-handler:inbound.springCloudBus.anonymous.f8fc9c0c-ccd3-46dd-9537-07198f4ee216} as a subscriber to the 'bridge.springCloudBus' channel
2016-09-28 22:50:39.163  INFO 828 --- [           main] o.s.c.s.b.k.KafkaMessageChannelBinder$7  : started inbound.springCloudBus.anonymous.f8fc9c0c-ccd3-46dd-9537-07198f4ee216

可以看到，config-client启动时输出了类似的内容，他们都订阅了名为springCloudBus的Topic。

在启动了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配置内容。

从config-client的控制台中，我们可以看到如下内容：

2016-09-29 08:20:34.361  INFO 21256 --- [ kafka-binder-1] o.s.cloud.bus.event.RefreshListener      : Received remote refresh request. Keys refreshed [from]

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多租户电子商城系统

微信关注我们

原文链接：https://yq.aliyun.com/articles/684960

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

函数计算PHP 变量不存在无法引用和判断的解决方案

最近研究了函数计算，很不错的服务，总结一个php存在的一个问题，为以后的小白们避免同样的错误。 <?php use RingCentral\Psr7\Response; function initializer($context) { echo 'initializing' . PHP_EOL; } function handler($request, $context): Response{ //$body = $request->getBody()->getContents(); $queries = $request->getQueryParams(); //$method = $request->getMethod(); //$headers = $request->getHeaders(); //$path = $request->getAttribute("path"); //$requestURI = $request->getAttribute("requestURI"); //$clientIP = $request-&...

2019-01-07

580

一、Spring面试题 1、Spring 在ssm中起什么作用？Spring：轻量级框架作用：Bean工厂，用来管理Bean的生命周期和框架集成。两大核心： 1、IOC/DI(控制反转/依赖注入) ：把dao依赖注入到service层，service层反转给action层，Spring顶层容器为BeanFactory。 2、AOP：面向切面编程 2、Spring的事务？编程式事务管理：编程方式管理事务，极大灵活性，难维护。声明式事务管理：可以将业务代码和事务管理分离，用注解和xml配置来管理事务。 3、IOC 在项目中的作用？作用：Ioc解决对象之间的依赖问题，把所有Bean的依赖关系通过配置文件或注解关联起来，降低了耦合度。 4、Spring的配置文件中的内容？开启事务注解驱动事务管理器开启注解功能，并配置扫描包配置数据库配置SQL会话工厂，别名，映射文件不用编写Dao层的实现类 5、Spring下的注解？注册：@Controller @Service @Component 注入：@Autowired @Resource 请求地址：@RequestMapping 返回...

2019-01-07

546

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。