关键词:
/* * Kafka配置文件, 用于定义producer, consumer * */ public class KafkaProperties //定义端口号 public static final String ZK = "localhost:2181"; public static final String TOPIC = "hello_topic"; public static final String BROKER_LIST = "localhost:9092";
- 定义全局变量topic,producer(选择kafka.javaapi.producer包)
- 写构造函数,包括了:
- 外部传入topic
- 创建producer,需要传入ProducerConfig对象
- PC对象需要传入一些参数,用properties类(java.util包)来传入
- properties对象中需要为PC对象设置”metadata.broker.list" “serializer.class" "request.required.acks"
import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; import java.util.Properties; /* * Kafka生产者 * */ public class KafkaProducer extends Thread private String topic; //选择kafka.javaapi.producer private Producer<Integer, String> producer; //构造方法,传入topic,生成producer public KafkaProducer(String topic) this.topic = topic; //用properties设置ProducerConfig所需要的参数, 这是生成Producer的前提 //分别是broker_list, 序列化, 握手机制 Properties properties = new Properties(); properties.put("metadata.broker.list", KafkaProperties.BROKER_LIST); properties.put("serializer.class", "kafka.serializer.StringEncoder"); //此处序列化类用String properties.put("request.required.acks", "1"); //可设置为0, 1, -1, 一般生产用1, 最严谨是-1, 不能用0 producer = new Producer<Integer, String>(new ProducerConfig(properties)); //用线程来启动producer @Override public void run() int messageNo = 1; while(true) String message = "massage_" + messageNo; producer.send(new KeyedMessage<Integer, String>(topic, message)); System.out.println("send: " + message); messageNo++; //2s间隔发送一次 try Thread.sleep(2000); catch(Exception e) e.printStackTrace();
- 构造方法中传入topic
- 创建createConnector方法,返回值是一个ConsumerConnector,注意不直接是Consumer
- 按照producer一样的方法,往ConsumerConnector中传入所需要的属性zookeeper.connect group.id
- 为了创建messageStream,先创建一个Map,装topic和kafka stream的数量
- 创建messageStream,并获取每次的数据
- 对messageStream进行迭代,获取消息
import kafka.consumer.Consumer; import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; import static com.sun.org.apache.xml.internal.security.keys.keyresolver.KeyResolver.iterator; import static javafx.scene.input.KeyCode.V; /* * Kafka消费者 * */ public class KafkaConsumer extends Thread private String topic; public KafkaConsumer(String topic) this.topic = topic; //ConsumerConnector选择kafka.javaapi.consumer包 //此处是要创建consumer连接器, 而不是创建consumer, 区别于producer private ConsumerConnector createConnector() //同样地设置ConsumerConfig对象的属性 //需要设置ZK Properties properties = new Properties(); properties.put("zookeeper.connect", KafkaProperties.ZK); properties.put("group.id", KafkaProperties.GROUP_ID); return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties)); //线程启动consumer @Override public void run() ConsumerConnector consumer = createConnector(); //由于createMessageStreams需要传入一个Map, 所以创建一个 Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); //map中放入topic和kafka stream的数量 topicCountMap.put(topic, 1); //创建messageStream, 从源码中可看出它的数据类型 //String是topic, List是数据比特流 Map<String, List<KafkaStream<byte[], byte[]>>> messageStream = consumer.createMessageStreams(topicCountMap); //获取每次的数据 KafkaStream<byte[], byte[]> byteStream = messageStream.get(topic).get(0); //数据流进行迭代 ConsumerIterator<byte[], byte[]> iterator = byteStream.iterator(); while (iterator.hasNext()) //由于iterator里面的是byte类型,要转为String String message = new String(iterator.next().message()); System.out.println("receive:" + message);
对countdownlatch的初步学习(代码片段)
CountDownLatch的中文翻译为"闭锁",在JDK1.5中CountDownLatch类加入进来。为程序猿进行并发编程提供有利的帮助。 首先我们先看看JDK文档中对于CountDownLatch类的介绍:Asynchronizationaidthatallowsoneormorethreadstowaituntilasetofoperationsbeingperforme... 查看详情
echarts初步学习(代码片段)
官方教程学习: 转自官网http://echarts.baidu.com/ ECharts3开始不再强制使用AMD的方式按需引入,代码里也不再内置AMD加载器。因此引入方式简单了很多,只需要像普通的JavaScript库一样用script标签引入。<!DOCTYPEhtml><html>... 查看详情
jsmap()初步学习(代码片段)
//array.map(callback,thisObject?),callback需要有return值//map:‘映射‘被映射成新的数组 eg1:letdata=[3,4,2];letdata2=data.map(item=>returnitem*5)console.log(data2) //eg2:letusers=[name:"天海佑希","em 查看详情
kafka学习之路kafka的安装(代码片段)
zookeeper1:192.168.1.11zookeeper2:192.168.1.12zookeeper3:192.168.1.13kafka1:192.168.1.14kafka2:192.168.1.15kafka3:192.168.1.16kafka3:192.168.1.17kafka-manager:192.168.1.18一、下载下载地址:http://kafka.apache. 查看详情
kafka学习笔记(代码片段)
安装kafka下载下载window的kafka地址window的kafka只是为了方便学习安装地址:kafka.apache.org/安装解压zip为文件夹启动kafkakafka服务器的功能相当于RocketMQ中的broker,kafka运行还需要一个类似于命名服务器的服务。在kafka安装目录... 查看详情
kafka学习之路kafka的架构(代码片段)
一、Kafka的集群架构如上图所示,一个典型的Kafka集群中包含若干Producer(可以是web前端产生的PageView,或者是服务器日志,系统CPU、Memory等),若干broker(Kafka支持水平扩展,一般broker数量越多,集群吞吐率越高),若干ConsumerGro... 查看详情
kafka集群搭建(代码片段)
...或复制Kafka安装包2解压缩Kafka安装包3配置Kafka集群4Kafka的初步应用4.1创建主题4.2发送消息4.3消费消息1下载或复制Kafka安装包http://archive.apache.org/dist/kafka/0.10.1.0/2解压缩Kafka安装包tar-zxvfkafka_2.11-0.10.1.0.gz3配置Kafka集群配置Kafka集群时... 查看详情
3.kafka学习(代码片段)
kafka是什么Kafka最初由Linkedin公司开发,是一个分布式、分区的、多副本的、多订阅者,基于zookeeper协调的,发布/订阅模式的消息队列(MessageQueue),Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。经过多年发展,Kafka已... 查看详情
centos初步学习fromlanguagec(代码片段)
一、Start1、编译器安装yuminstallgcc-y#c编译器yuminstallgcc-c++-y#c++编译器2、第一个C程序//文件名:txl.c#include<stdio.h>//标准输出输入intmain()printf(“hello”);return0;3、编译gcctxl.c-otxl-o:编译并输出一个可执行文件 查看详情
消息系统概述(代码片段)
草捏子最近开始学习Kafka。在学习前,给自己确定了下学习的范围,大致如下:理解Kafka的相关概念;掌握Kafka的基本API使用;了解Kafka的背后原理。后续将在这学习范围内输出一些相关文章。那么本文作为Kafka系列的第一篇文章... 查看详情
storm初步(代码片段)
初入Storm前言学习Storm已经有两周左右的时间,但是认真来说学习过程确实是零零散散,遇到问题去百度一下,找到新概念再次学习,在这样的一个循环又不成体系的过程中不断学习Storm。前人栽树,后人乘凉,也正是因为网上有... 查看详情
docker了解及初步学习(代码片段)
部署项目是一件很麻烦的事,尤其是当同一个项目部署到很多服务器的时候,这种麻烦变得更加让人难受。 在我刚到公司的时候,公司之前的先项目采用的就是使用Web服务器Tomcat部署项目,但是后来的新项目是微服务的... 查看详情
kafka学习入门与原理(代码片段)
Kafka学习(一)入门与原理初识KafkaKafka起初是由LinkedIn公司采用Scala语言开发的一个多分区、多副本且基于ZooKeeper协调的分布式消息系统,现已被捐献给Apache基金会。基本概念一个典型的Kafka体系架构包括若干Producer、若干Broker、... 查看详情
日志打入kafka改造历程-我们到底能走多远系列49(代码片段)
...。另一方面,应用在服务器上会打日志文件。如图:详细初步实现首先,我们来初步实现这个方案,搭建elk略去不谈,其中特 查看详情
kafka学习之路kafka在zookeeper中的存储(代码片段)
当kafka启动的时候,就会向zookeeper里面注册一些信息,这些数据也称为Kafka的元数据信息。一、Kafka在zookeeper中存储结构图二、分析根目录下的结构服务端开启的情况下,进入客户端的命令:zookeeper目录/bin/zkCli.sh#zookeeper目录/bin/zk... 查看详情
kafka学习笔记&面经分享(代码片段)
目录定义使用消息队列的好处消息队列的两种模式点对点模式发布/订阅模式kafka基础架构kafka架构kafka工作流程kafka文件存储机制kafka生产者分区策略分区的原因分区的原则数据可靠性保证副本数据同步策略ISRack应答机制故障处理... 查看详情
机器学习sklearn的k近邻算法api初步使用(代码片段)
目录1k近邻算法api初步使用2Scikit-learn工具介绍2.1安装2.2Scikit-learn包含的内容3K-近邻算法API4案例4.1步骤分析4.2代码过程5小结1k近邻算法api初步使用K近邻算法介绍:https://blog.csdn.net/ZGL_cyy/article/details/125583129机器学习流程复习... 查看详情
vue学习之路-1.初步感知(代码片段)
一、安装 这里使用node的npm包管理工具进行操作。操作前请先下载node。 在工程文件夹中使用以下命令安装vue:npminstallvue 如下图所示:我在helloworld 文件夹中安装vue: // Vue不支持IE8及其以... 查看详情