当前位置： 首页 > news >正文

网站建设需要微信账号和密码高端网站建设公司的问题

news 2025/9/16 12:07:38
网站建设需要微信账号和密码,高端网站建设公司的问题,搜索广告排名,小程序开发问题1.环境准备： 使用如下3台主机搭建zookeeper集群，由于默认的9092客户端连接端口不在本次使用的云服务器开放端口范围内，故端口改为了8093。 172.2.1.69:8093 172.2.1.70:8093 172.2.1.71:8093 2.下载地址 去官网下载，或者使用如…

1.环境准备：

使用如下3台主机搭建zookeeper集群，由于默认的9092客户端连接端口不在本次使用的云服务器开放端口范围内，故端口改为了8093。
172.2.1.69:8093
172.2.1.70:8093
172.2.1.71:8093

2.下载地址

去官网下载，或者使用如下仓库地址下载，本次使用的时kafka_2.13-3.6.1.tgz ，即3.6.1版本，前面的2.13是scala版本，该版本是较新的版本，可以使用zookeeper，也可以不使用zookeeper搭建集群，本次记录使用了zk，zk集群的部署可以参考上一篇记录。

# 软件包下载地址，可以切到/kafka/路径，选择自己需要的版本
https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgz

3.软件包下载解压

在上面3台服务器上分别执行wget下载，或者本地下载后上传，本次使用的环境为堡垒机接入，如果使用的是宿主机账密登陆，可以下载配置一台，其余使用SCP命令拷贝过去即可。

cd  /usr/local/wget https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgztar -zxvf kafka_2.13-3.6.1.tgzmv kafka_2.13-3.6.1 kafka

4.修改配置

需要修改logs路径的话，可以在/kafka路径下新建logs路径，并配置到server.properties中，这里使用默认的/tmp/kafka-logs路径。

cd kafka
vim conf/server.properties

修改内容如下，

# 每个节点唯一的id，这里.69、.70、.71服务器分别设置为了1、2、3
broker.id=1# 默认为9092，云服务器开放端口问题，改为了8093
port=8093# 上一篇记录博客搭建的zk集群地址
zookeeper.connect=172.2.1.69:8092,172.2.1.70:8092,172.2.1.71:8092# 配置监听访问、绑定地址，这里都是PLAINTEXT协议，不需要认证（相当于内网访问）
listeners=PLAINTEXT://0.0.0.0:8093                                                                                                 
advertised.listeners=PLAINTEXT://172.2.1.71:8093# 日志路径
log.dirs=/tmp/kafka-logs

5.启动kafka集群

分别在每个节点的bin路径下执行启动脚本

# 在3个节点分别执行如下命令，-daemon表示后台启动，不带该参数前台启动
./bin/kafka-server-start.sh -daemon config/server.properties

使用jps命令，或者去kafka启动日志，查看kafka是否启动成功。

6.创建topic`

# 创建topic，在任一节点执行都可以。./bin/kafka-topics.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --create --topic topic-demo --partitions=3 --replication-factor=3
# 查看topic是否创建成功，在任一节点执行
./bin/kafka-topics.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --topic topic-demo --list

7.模拟生产消费消息

需要注意的是网上搜到的一些老的博客kafka命令在高版本中是不再支持的，如：
sh ./bin/kafka-topics.sh --zookeeper=zk集群地址，可能出现命令无法识别：zookeeper is not a recognized option
需要替换为：sh ./bin/kafka-topics.sh --bootstrap-server=kafka集群地址，注意–bootstrap-server后面跟的是kafka集群地址，不是zookeeper地址。

# 在2个节点启动消费者模拟客户端接收消息，在第3个节点启动生产者模拟客户端发送消息
./bin/kafka-console-consumer.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --topic topic-demo

# 在第3个节点启动生产者客户端模拟发送消息：hello
./bin/kafka-console-producer.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --topic topic-demo

生产者客户端发送hello
在这里插入图片描述
此时可以看到2个消费者模拟客户端都受到了消息：hello
在这里插入图片描述

8.集成springboot

坐标如下：

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

kafka配置类：

package com.example.kafka.kafka;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;import java.util.HashMap;
import java.util.Map;@Configuration
@EnableKafka
//@RefreshScope
public class KafkaConfig {@Value("${xxxx:172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093}")private String kafkaServers;public Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);props.put(ProducerConfig.RETRIES_CONFIG, 0);props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);props.put(ProducerConfig.LINGER_MS_CONFIG, 1);props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return props;}@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}public Map<String, Object> consumerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);//props.put(ConsumerConfig.GROUP_ID_CONFIG, "0");props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "20000");props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);return props;}@Beanpublic ConsumerFactory<String, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs());}@BeanConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());return kafkaTemplate;}
}

消费者客户端：

    @KafkaListener(topics = {"topic-demo"},groupId = "test1",properties = {"auto-offset-reset:latest", "enable.auto.commit:true"})public void listen(ConsumerRecord<String, String> consumerRecord) {log.info("consumer Received: " + consumerRecord);}

生产者发送消息：

@RestController
@RequiredArgsConstructor
@RequestMapping("producer")
public class ProducerController {private final KafkaTemplate<String, String> kafkaTemplate;@PostMapping(path = "/sendCommonMsg")public String sendCommonMsg(String topic, String msg) {ListenableFuture<SendResult<String, String>> hello_kafka = this.kafkaTemplate.send(topic, "hello kafka");SendResult<String, String> sendResult = hello_kafka.completable().join();System.out.println(sendResult);return "send topic: " + topic + ", msg: " + msg;}
}

发送测试：
在这里插入图片描述

消费者可以接收到消息：

consumer Received: ConsumerRecord(topic = topic-demo, partition = 0, leaderEpoch = 0, offset = 2, CreateTime = 1721720091071, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello)

9.IDEA客户端工具

可以使用kafkalytic工具本地开发环境可视化操作kafka服务器，如查看topic，创建topic

在这里插入图片描述

http://www.yayakq.cn/news/219693/

相关文章：

  • 哈尔滨网站建设培训王也天与葛优
  • 西安哪家做网站最好短网址生成功能
  • html5 手机网站导航条劳务公司网站建设
  • 做哪个网站的推广最好wordpress个人中心
  • 应用数据库网站开发找第三方做网站 需要注意
  • 外贸网站导航国税部门强化网站建设
  • 杨凌做网站的公司百度资源共享
  • windows iis运行php网站打广告推广怎么做
  • 泰州网站制作计划网站微信建设方案
  • 厦门网站建设阳哥上海华东建设发展设计有限公司网站
  • 关于服装的商务网站建设策划书wordpress 搭建
  • 郑州网站外包公司sae wordpress 4.3
  • 保定外贸网站制作硬件开发专业
  • 关注网站建设公司名被注册网站
  • 优购物官方网站购物wordpress手机全部显示
  • 全球50个大网站开发语言低价格的网站建设公司
  • 公司网站制作设计联系方式企业解决方案有哪些
  • 微信公众号的微网站怎么做的遵义市在哪里做网站
  • 微信网站开发完全教程看男女做那个视频网站
  • 建设c2c网站需要多少投资科技网站首页欣赏
  • 所有网站名称大全三门峡市住房建设局网站
  • 保定电商网站建设手工艺品网站建设方案
  • 打鱼网站开发梅州建站规划
  • wordpress什么协议论坛seo设置
  • 常德 网站建设做网站公众号多少钱
  • 嘉兴企业网站设计哪家好wordpress get_search_form
  • 网站建设专业可行性分析网站做app的软件叫什么
  • 广州培训网站开发古建设工程造价管理协会网站
  • 绍兴模板建站公司软件工网站开发课程设计报告
  • 高端网站建设磐石网络好吴江做网站公司