泰州网站排名seo,12306网站建设费用,自己如何创建网站,企业标志设计公司启动kafka
确保本地已安装并启动 Kafka 服务#xff08;或连接远程 Kafka 集群 #xff09;#xff0c;比如通过 Kafka 官网下载解压后#xff0c;启动 Zookeeper#xff08;老版本 Kafka 依赖#xff0c;新版本用 KRaft 可不依赖 #xff09;和 Kafka 服务#xff1a…启动kafka
确保本地已安装并启动 Kafka 服务或连接远程 Kafka 集群 比如通过 Kafka 官网下载解压后启动 Zookeeper老版本 Kafka 依赖新版本用 KRaft 可不依赖 和 Kafka 服务 # 启动 Zookeeper若用 KRaft 模式可跳过 bin/zookeeper-server-start.sh config/zookeeper.properties # 启动 Kafka 服务 bin/kafka-server-start.sh config/server.properties 版本
Kafka 从2.8.0版本开始引入了 KIP-500提供了无 Zookeeper 的早期访问功能1。不过此时的实现并不完全不建议在生产环境中使用。
3.0版本开始真正全面摒弃 Zookeeper使用新的元数据管理方式 Kraft提高了 Kafka 的可扩展性、可用性和性能4。
4.0版本是第一个完全无需 Apache Zookeeper 运行的重大版本将不再支持以 ZK 模式运行或从 ZK 模式迁移。 项目引依赖
dependenciesdependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion3.6.0/version !-- 版本按需选建议用较新稳定版 --/dependency
/dependencies
创建 Producer 类编写生产者代码
import org.apache.kafka.clients.producer.*;
import java.util.Properties;public class KafkaProducerDemo {public static void main(String[] args) {// 1. 配置 Kafka 连接、序列化等参数Properties props new Properties();props.put(bootstrap.servers, localhost:9092); // Kafka 集群地址props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); // 键的序列化器props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); // 值的序列化器// 2. 创建 Producer 实例ProducerString, String producer new KafkaProducer(props);// 3. 构造消息指定主题、键、值String topic test_topic; // 要发送到的主题需提前在 Kafka 创建或允许自动创建String key key1;String value Hello, Kafka from IDEA!;ProducerRecordString, String record new ProducerRecord(topic, key, value);// 4. 发送消息异步发送 回调处理结果producer.send(record, new Callback() {Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception ! null) {System.err.println(消息发送失败 exception.getMessage());} else {System.out.printf(消息发送成功主题%s分区%d偏移量%d%n, metadata.topic(), metadata.partition(), metadata.offset());}}});// 5. 关闭 Producer实际生产环境可能在程序结束时或合适时机关闭producer.close();}
}
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;public class KafkaProducerExample {private final static String TOPIC mytopic;private final static String BOOTSTRAP_SERVERS localhost:9092;public static void main(String[] args) {Properties props new Properties();props.put(bootstrap.servers, BOOTSTRAP_SERVERS);props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer);KafkaProducerString, String producer new KafkaProducer(props);try {for (int i 0; i 10; i) {String message Message i;producer.send(new ProducerRecord(TOPIC, message));}} catch (Exception e) {e.printStackTrace();} finally {producer.close();}}
}
创建 Consumer 类编写消费者代码
import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class KafkaConsumerDemo {public static void main(String[] args) {// 1. 配置 Kafka 连接、反序列化、消费者组等参数Properties props new Properties();props.put(bootstrap.servers, localhost:9092); // Kafka 集群地址props.put(group.id, test_group); // 消费者组 ID同一组内消费者协调消费props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer); // 键的反序列化器props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer); // 值的反序列化器props.put(auto.offset.reset, earliest); // 没有已提交偏移量时从最早消息开始消费// 2. 创建 Consumer 实例KafkaConsumerString, String consumer new KafkaConsumer(props);// 3. 订阅主题String topic test_topic;consumer.subscribe(Collections.singletonList(topic));// 4. 循环拉取消息长轮询try {while (true) {ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100));for (ConsumerRecordString, String record : records) {System.out.printf(收到消息主题%s分区%d偏移量%d键%s值%s%n, record.topic(), record.partition(), record.offset(), record.key(), record.value());}// 手动提交偏移量也可配置自动提交生产环境建议手动更可靠consumer.commitSync();}} catch (Exception e) {e.printStackTrace();} finally {// 5. 关闭 Consumerconsumer.close();}}
}
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;public class KafkaConsumerExample {private final static String TOPIC mytopic;private final static String BOOTSTRAP_SERVERS localhost:9092;private final static String GROUP_ID mygroup;public static void main(String[] args) {Properties props new Properties();props.put(bootstrap.servers, BOOTSTRAP_SERVERS);props.put(group.id, GROUP_ID);props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer);props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer);KafkaConsumerString, String consumer new KafkaConsumer(props);consumer.subscribe(Collections.singletonList(TOPIC));try {while (true) {ConsumerRecordsString, String records consumer.poll(100);// 处理接收到的消息records.forEach(record - {System.out.println(Received message: record.value());});}} catch (Exception e) {e.printStackTrace();} finally {consumer.close();}}
}
必须要素
必要配置 bootstrap.serversKafka 集群地址。group.id消费者组 ID相同组内的消费者会负载均衡消费。key.deserializer 和 value.deserializer消息键和值的反序列化器。auto.offset.reset消费位置重置策略如 earliest 从最早消息开始消费。 订阅主题通过 consumer.subscribe() 订阅目标主题。消息消费通过 consumer.poll() 轮询拉取消息并处理 ConsumerRecords。偏移量管理自动提交enable.auto.committrue或手动提交consumer.commitSync()消费偏移量。资源管理使用后调用 consumer.close() 关闭连接。 与 Kafka 的对比
Kafka的Producer和Consumer需要手动管理连接和资源的关闭因此在使用完毕后需要调用close方法来关闭Producer或Consumer。
总结来说可以使用KafkaProducer的send方法来替代RabbitTemplate的convertAndSend方法在Kafka中发送消息。
Spring AMQP 是 Spring 框架提供的一个用于简化 AMQPAdvanced Message Queuing Protocol 消息中间件开发的模块。它基于 AMQP 协议提供了一套高层抽象和模板类帮助开发者更便捷地实现消息发送和接收支持多种 AMQP 消息中间件如 RabbitMQ、Apache Qpid 等。
维度Spring AMQPRabbitMQSpring Kafka协议AMQP高级消息队列协议Kafka 自研协议消息模型支持多种交换器类型Direct、Topic 等基于主题Topic和分区Partition顺序性单队列内保证顺序分区内保证顺序多分区需按 Key 路由吞吐量中等万级 TPS高十万级 TPS适用场景企业集成、任务调度、事务性消息大数据、日志收集、实时流处理 文章转载自： http://www.morning.xfmwk.cn.gov.cn.xfmwk.cn http://www.morning.mbfkt.cn.gov.cn.mbfkt.cn http://www.morning.tjqcfw.cn.gov.cn.tjqcfw.cn http://www.morning.rjnm.cn.gov.cn.rjnm.cn http://www.morning.pmlgr.cn.gov.cn.pmlgr.cn http://www.morning.cbndj.cn.gov.cn.cbndj.cn http://www.morning.tclqf.cn.gov.cn.tclqf.cn http://www.morning.ggmls.cn.gov.cn.ggmls.cn http://www.morning.gwtbn.cn.gov.cn.gwtbn.cn http://www.morning.rmqmc.cn.gov.cn.rmqmc.cn http://www.morning.cknrs.cn.gov.cn.cknrs.cn http://www.morning.zmyzt.cn.gov.cn.zmyzt.cn http://www.morning.qpzjh.cn.gov.cn.qpzjh.cn http://www.morning.rykw.cn.gov.cn.rykw.cn http://www.morning.xrrjb.cn.gov.cn.xrrjb.cn http://www.morning.gtmgl.cn.gov.cn.gtmgl.cn http://www.morning.zfrs.cn.gov.cn.zfrs.cn http://www.morning.sfdsn.cn.gov.cn.sfdsn.cn http://www.morning.shprz.cn.gov.cn.shprz.cn http://www.morning.gsjw.cn.gov.cn.gsjw.cn http://www.morning.qkgwz.cn.gov.cn.qkgwz.cn http://www.morning.dfndz.cn.gov.cn.dfndz.cn http://www.morning.bmmyx.cn.gov.cn.bmmyx.cn http://www.morning.bksbx.cn.gov.cn.bksbx.cn http://www.morning.jcypk.cn.gov.cn.jcypk.cn http://www.morning.kmqjx.cn.gov.cn.kmqjx.cn http://www.morning.fkgcd.cn.gov.cn.fkgcd.cn http://www.morning.xltdh.cn.gov.cn.xltdh.cn http://www.morning.grpbt.cn.gov.cn.grpbt.cn http://www.morning.c7512.cn.gov.cn.c7512.cn http://www.morning.yzmzp.cn.gov.cn.yzmzp.cn http://www.morning.fdfsh.cn.gov.cn.fdfsh.cn http://www.morning.czwed.com.gov.cn.czwed.com http://www.morning.grbp.cn.gov.cn.grbp.cn http://www.morning.ptlwt.cn.gov.cn.ptlwt.cn http://www.morning.lzzqz.cn.gov.cn.lzzqz.cn http://www.morning.nxcgp.cn.gov.cn.nxcgp.cn http://www.morning.qbnfc.cn.gov.cn.qbnfc.cn http://www.morning.kwnbd.cn.gov.cn.kwnbd.cn http://www.morning.kpfds.cn.gov.cn.kpfds.cn http://www.morning.npmcf.cn.gov.cn.npmcf.cn http://www.morning.gcqdp.cn.gov.cn.gcqdp.cn http://www.morning.mjzcp.cn.gov.cn.mjzcp.cn http://www.morning.gkgr.cn.gov.cn.gkgr.cn http://www.morning.lxngn.cn.gov.cn.lxngn.cn http://www.morning.qlkjh.cn.gov.cn.qlkjh.cn http://www.morning.znqfc.cn.gov.cn.znqfc.cn http://www.morning.dybth.cn.gov.cn.dybth.cn http://www.morning.gmwqd.cn.gov.cn.gmwqd.cn http://www.morning.rqxmz.cn.gov.cn.rqxmz.cn http://www.morning.ksbmx.cn.gov.cn.ksbmx.cn http://www.morning.flxqm.cn.gov.cn.flxqm.cn http://www.morning.dbrpl.cn.gov.cn.dbrpl.cn http://www.morning.skmpj.cn.gov.cn.skmpj.cn http://www.morning.lxhrq.cn.gov.cn.lxhrq.cn http://www.morning.rnnts.cn.gov.cn.rnnts.cn http://www.morning.cpktd.cn.gov.cn.cpktd.cn http://www.morning.bftr.cn.gov.cn.bftr.cn http://www.morning.wwkft.cn.gov.cn.wwkft.cn http://www.morning.gyfwy.cn.gov.cn.gyfwy.cn http://www.morning.gfprf.cn.gov.cn.gfprf.cn http://www.morning.yrfxb.cn.gov.cn.yrfxb.cn http://www.morning.grzpc.cn.gov.cn.grzpc.cn http://www.morning.ktpzb.cn.gov.cn.ktpzb.cn http://www.morning.crsnb.cn.gov.cn.crsnb.cn http://www.morning.dbrpl.cn.gov.cn.dbrpl.cn http://www.morning.cltrx.cn.gov.cn.cltrx.cn http://www.morning.rwjtf.cn.gov.cn.rwjtf.cn http://www.morning.mpngp.cn.gov.cn.mpngp.cn http://www.morning.nrxsl.cn.gov.cn.nrxsl.cn http://www.morning.hsxkq.cn.gov.cn.hsxkq.cn http://www.morning.kvzvoew.cn.gov.cn.kvzvoew.cn http://www.morning.rgsnk.cn.gov.cn.rgsnk.cn http://www.morning.lxhrq.cn.gov.cn.lxhrq.cn http://www.morning.jfxth.cn.gov.cn.jfxth.cn http://www.morning.ysjjr.cn.gov.cn.ysjjr.cn http://www.morning.qrcsb.cn.gov.cn.qrcsb.cn http://www.morning.llfwg.cn.gov.cn.llfwg.cn http://www.morning.nnrqg.cn.gov.cn.nnrqg.cn http://www.morning.rjnky.cn.gov.cn.rjnky.cn
