网站上怎么做游戏,深圳小语种网站建设,宁夏住宅建设发展公司网站,东莞保安招聘网文章目录一、消费者监听1. 启动zk2. 启动kafka3. 创建主题4. 消费者监听消息二、生产者工程2.1. 依赖2.2. 生产者代码#xff08;同步#xff09;2.3. 生产者代码#xff08;异步#xff09;2.4. 发送消息2.5. 消费者监听消息2.6. 结果返回一、消费者监听
1. 启动zk
zkSe…
文章目录一、消费者监听1. 启动zk2. 启动kafka3. 创建主题4. 消费者监听消息二、生产者工程2.1. 依赖2.2. 生产者代码同步2.3. 生产者代码异步2.4. 发送消息2.5. 消费者监听消息2.6. 结果返回一、消费者监听
1. 启动zk
zkServer.sh start# 监听运行状态
zkServer.sh status2. 启动kafka
# 后台启动kafka
kafka-server-start.sh -daemon /app/kafka_2.12-2.8.0/config/server.properties3. 创建主题
# 创建一个主题名称为topic_1 该主题分区1个分区 该分区有1个副本
kafka-topics.sh --zookeeper localhost:2181/mykafka --create --topic topic_1 --partitions 1 --replication-factor 14. 消费者监听消息
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1二、生产者工程
2.1. 依赖 dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion2.7.1/version/dependency2.2. 生产者代码同步
package com.gblfy.kafka.producer;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class MyProducer1 {public static void main(String[] args) throws ExecutionException, InterruptedException {MapString, Object configs new HashMapString, Object();//指定初始化连接用到的broker地址configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.159.102:9092);//指定key序列化类configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);//指定value序列化类configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);//构造生产者对象 指定发送的key和value的类型 配置的参数列表必填参数辅助参数KafkaProducerInteger, String producer new KafkaProducerInteger, String(configs);//用于设置用户自定义的消息头字段ListHeader headers new ArrayList();headers.add(new RecordHeader(biz.name, producer.demo.getBytes()));//构造record封装发送消息主体ProducerRecordInteger, String record new ProducerRecordInteger, String(topic_1, //指定发送主题0,//指定发送分区0,//指定发送keyhello gblfy 0,//指定发送消息主题headers//用于设置用户自定义的消息头字段);//消息的同步确认 调用send方法发送消息final FutureRecordMetadata future producer.send(record);//调用get方法接收消息final RecordMetadata metadata future.get();System.out.println(消息的主题: metadata.topic());System.out.println(消息的分区: metadata.partition());System.out.println(消息的偏移量: metadata.offset());//关闭生产者producer.close();}
}
2.3. 生产者代码异步
package com.gblfy.kafka.producer;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class MyProducer1 {public static void main(String[] args) throws ExecutionException, InterruptedException {MapString, Object configs new HashMapString, Object();//指定初始化连接用到的broker地址configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.159.102:9092);//指定key序列化类configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);//指定value序列化类configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);//构造生产者对象 指定发送的key和value的类型 配置的参数列表必填参数辅助参数KafkaProducerInteger, String producer new KafkaProducerInteger, String(configs);//用于设置用户自定义的消息头字段ListHeader headers new ArrayList();headers.add(new RecordHeader(biz.name, producer.demo.getBytes()));//构造record封装发送消息主体ProducerRecordInteger, String record new ProducerRecordInteger, String(topic_1, //指定发送主题0,//指定发送分区0,//指定发送keyhello gblfy 0,//指定发送消息主题headers//用于设置用户自定义的消息头字段);//消息的同步确认 调用send方法发送消息final FutureRecordMetadata future producer.send(record);//调用get方法接收消息// final RecordMetadata metadata future.get();// System.out.println(消息的主题: metadata.topic());// System.out.println(消息的分区: metadata.partition());// System.out.println(消息的偏移量: metadata.offset());//消息的异步确认producer.send(record, new Callback() {Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception null) {System.out.println(消息的主题: metadata.topic());System.out.println(消息的分区: metadata.partition());System.out.println(消息的偏移量: metadata.offset());} else {System.out.println(异常消息);}}});//关闭生产者producer.close();}
}
2.4. 发送消息 消息有同步发送和异步发送二种 2.5. 消费者监听消息 2.6. 结果返回
