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

西宁电商网站建设建设建材网站的目的

西宁电商网站建设,建设建材网站的目的,企业vi手册范本,江苏省建设考试信息管理系统网站这篇文章简单介绍如何在ubuntu上安装kafka#xff0c;并使用kafka完成消息的发送和接收。 一、安装kafka 访问kafka官网Apache Kafka#xff0c;然后点击快速开始 紧接着#xff0c;点击Download 最后点击下载链接下载安装包 如果下载缓慢#xff0c;博主已经把安装包上传… 这篇文章简单介绍如何在ubuntu上安装kafka并使用kafka完成消息的发送和接收。 一、安装kafka 访问kafka官网Apache Kafka然后点击快速开始 紧接着点击Download 最后点击下载链接下载安装包 如果下载缓慢博主已经把安装包上传到百度网盘 链接https://pan.baidu.com/s/1nZ1duIt64ZVUsimaQ1meZA?pwd3aoh 提取码3aoh --来自百度网盘超级会员V3的分享 二、启动kafka 经过上一步下载完成后按照页面的提示启动kafka 1、通过远程连接工具如finalshell、xshell上传kafka_2.13-3.6.0.tgz到服务器上的usr目录 2、切换到usr目录解压kafka_2.13-3.6.0.tgz cd /usrtar -zxzf kafka_2.13-3.6.0.tgz 3、启动zookeeper 修改配置文件confg/zookeeper.properties修改一下数据目录 dataDir/usr/local/zookeeper 然后通过以下命令启动kafka自带的zookeeper bin/zookeeper-server-start.sh config/zookeeper.properties 4、启动kafka 修改配置文件confg/server.properties修改一下kafka保存日志的目录 log.dirs/usr/local/kafka/logs 然后新开一个连接窗口通过以下命令启动kafka bin/kafka-server-start.sh config/server.properties 三、kafka发送、接收消息 创建topic bin/kafka-topics.sh --create --topic hello --bootstrap-server localhost:9092 生产消息 往刚刚创建的topic里发送消息可以一次性发送多条消息点击CtrlC完成发送 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic hello 消费消息 消费最新的消息 新开一个连接窗口在命令行输入以下命令拉取topic为hello上的消息 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello 消费之前的消息 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic hello 指定偏移量消费 指定从第几条消息开始消费这里--offset参数设置的偏移量是从0开始的。 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --partition 0 --offset 1 --topic hello 消息的分组消费 每个消费者都可以指定一个消费者组 kafka 中的同一条消息只能被同一个消费者组下的某一个消费 者消费。而不属于同一个消费者组的其他消费者也可以消费到这一条消息。 通过以下命令在启动消费者时设置分组 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.idhelloGroup --topic hello 四、Java中使用kafka 通过maven官网搜索kafka的maven依赖版本 https://central.sonatype.com/search?qkafkahttps://central.sonatype.com/search?qkafka然后通过IntelliJ IDEA创建一个maven项目kafka在pom.xml中添加kafka的依赖 ?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdorg.example/groupIdartifactIdkafka/artifactIdversion1.0-SNAPSHOT/versionpropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.target/propertiesdependenciesdependencygroupIdorg.apache.kafka/groupIdartifactIdkafka_2.12/artifactIdversion3.6.0/version/dependency/dependencies /project 创建消息生产者 生产者工厂类 package producer;import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig;import java.util.Properties;/*** 消息生产者工厂类* author heyunlin* version 1.0*/ public class MessageProducerFactory {private static final String BOOTSTRAP_SERVERS 192.168.254.128:9092;public static ProducerString, String getProducer() {//PART1:设置发送者相关属性Properties props new Properties();// 此处配置的是kafka的端口props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);// 配置key的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);// 配置value的序列化类props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);return new KafkaProducer(props);}} 测试发送消息 package producer;import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata;/*** author heyunlin* version 1.0*/ public class MessageProducer {private static final String TOPIC hello;public static void main(String[] args) {ProducerRecordString, String record new ProducerRecord(TOPIC, 1, Message From Producer.);ProducerString, String producer MessageProducerFactory.getProducer();// 同步发送消息producer.send(record);// 异步发送消息producer.send(record, new Callback() {Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {String topic recordMetadata.topic();long offset recordMetadata.offset();int partition recordMetadata.partition();String message recordMetadata.toString();System.out.println(topic topic);System.out.println(offset offset);System.out.println(message message);System.out.println(partition partition);}});// 加上这行代码才会发送消息producer.close();}} 创建消息消费者 消费者工厂类 package consumer;import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer;import java.util.Properties;/*** 消息生产者工厂类* author heyunlin* version 1.0*/ public class MessageConsumerFactory {private static final String BOOTSTRAP_SERVERS 192.168.254.128:9092;public static ConsumerString, String getConsumer() {//PART1:设置发送者相关属性Properties props new Properties();//kafka地址props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);//每个消费者要指定一个groupprops.put(ConsumerConfig.GROUP_ID_CONFIG, helloGroup);//key序列化类props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);//value序列化类props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);return new KafkaConsumer(props);}} 测试消费消息 package consumer;import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords;import java.time.Duration; import java.util.Collections;/*** author heyunlin* version 1.0*/ public class MessageConsumer {private static final String TOPIC hello;public static void main(String[] args) {ConsumerString, String consumer MessageConsumerFactory.getConsumer();consumer.subscribe(Collections.singletonList(TOPIC));while (true) {ConsumerRecordsString, String records consumer.poll(Duration.ofNanos(100));for (ConsumerRecordString, String record : records) {System.out.println(record.key() : record.value());}// 提交偏移量避免消息重复推送consumer.commitSync(); // 同步提交// consumer.commitAsync(); // 异步提交}}} 五、springboot整合kafka 开始前的准备工作 然后通过IntelliJ IDEA创建一个springboot项目springboot-kafka在pom.xml中添加kafka的依赖 dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId /dependency 然后修改application.yml添加kafka相关配置 spring:kafka:bootstrap-servers: 192.168.254.128:9092producer:acks: 1retries: 3batch-size: 16384properties:linger:ms: 0buffer-memory: 33554432key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: helloGroupenable-auto-commit: falseauto-commit-interval: 1000auto-offset-reset: latestproperties:request:timeout:ms: 18000session:timeout:ms: 12000key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer 创建消息生产者 package com.example.springboot.kafka.producer;import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController;/*** author heyunlin* version 1.0*/ RestController RequestMapping(path /producer, produces application/json;charsetutf-8) public class KafkaProducer {private final KafkaTemplateString, Object kafkaTemplate;Autowiredpublic KafkaProducer(KafkaTemplateString, Object kafkaTemplate) {this.kafkaTemplate kafkaTemplate;}RequestMapping(value /sendMessage, method RequestMethod.GET)public String sendMessage(String message) {kafkaTemplate.send(hello, message);return 发送成功~;}} 创建消息消费者 package com.example.springboot.kafka.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;/*** author heyunlin* version 1.0*/ Component public class KafkaConsumer {KafkaListener(topics hello)public void receiveMessage(ConsumerRecordString, String record) {String topic record.topic();long offset record.offset();int partition record.partition();System.out.println(topic topic);System.out.println(offset offset);System.out.println(partition partition);}} 然后访问网址http://localhost:8080/producer/sendMessage?messagehello往topic为hello的消息队列发送消息。控制台打印了参数成功监听到发送的消息。 文章涉及的项目已经上传到gitee按需获取~ Java中操作kafka的基本项目https://gitee.com/he-yunlin/kafka.git springboot整合kafka案例项目https://gitee.com/he-yunlin/springboot-kafka.git
http://www.w-s-a.com/news/479543/

相关文章:

  • 制作深圳网站建设阿里OSS做网站图库费用
  • 网页设计与网站建设 入门必练宜都网站seo
  • 网站设计沟通阆中网站网站建设
  • 缩短网址做钓鱼网站如何确保网站安全
  • 网店网站开发怎样用ps做企业网站
  • 南京门户网站建设做网站一般注册哪几类商标
  • 企业咨询管理服务wordpress seo tdk
  • 做网站前期创建文件夹flash 开发的网站
  • 天津网站制作培训搭建网站的工具
  • 江西网站建设价格低网上卖产品怎么推广
  • 做aelogo动效有什么好的网站建立网站站点的步骤
  • 无锡网站推广公司排名网站赚钱平台
  • 网站物理结构利用公共dns做网站解析
  • 苍南做网站微信建设网站哪家好
  • 华中农业大学基因编辑在线设计网站网站蓝色配色
  • 洛阳建设工程信息网站怎么在服务器建立网站
  • 陕西省建设部官方网站微信小程序怎么关闭这个功能
  • 唐河县住房和城乡建设局网站公司需要做网站
  • 体现网站特色免费个人域名网站
  • ps国外教程网站seo优化是什么职业
  • 什么是网站单页适合女生做的网站
  • 环境文化建设方案网站企业英语网站
  • 南通网站关键词推广响应式网站建设流程
  • 湖北响应式网站建设企业做漫画网站 漫画哪找
  • 东莞建设通网站中小企业网站的建设实践报告
  • 合肥网站建设电话wordpress 点击量
  • 公司网站制作注意什么wordpress如何邀请人看
  • 做渲染的网站太原做网站兼职
  • 网站开发实施方案怎么设置wordpress底栏文字
  • 网站建设朝阳学前端有必要找培训机构吗