`

Kafka单机环境开发示例

 
阅读更多

Kafka下载页面下载0.8版本,解压缩。

1.修改config目录下的server.properties 里面的host.name为机器的ip。假如部署kafka和开发运行kafka例子为同一台机器,不用修改,用默认的localhost也行。

2.修改config目录下的zookeeper.properties 里面的dataDir属性为你需要的目录。

3.假如你要配置集群,在kafka解压缩目录下新建zoo_data目录(第一次的时候需要新建),在zoo_data目录新建myid文件,设置内容为1。同时修改zookeeper.properties,具体可参考:solrcloud在tomcat下安装(三)

4.启动kafka。

//启动zookeeper server (用&是为了能退出命令行):
bin/zookeeper-server-start.sh config/zookeeper.properties  & 
//启动kafka server:  
bin/kafka-server-start.sh config/server.properties  &

5.新建一个生产者例子

import java.util.Properties;
 
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
 
public class KafkaTest {
	public static void main(String[] args) {  
        Properties props = new Properties();  
        props.put("zk.connect", "10.103.22.47:2181");  
        props.put("serializer.class", "kafka.serializer.StringEncoder");  
        props.put("metadata.broker.list", "10.103.22.47:9092");
        props.put("request.required.acks", "1");
        //props.put("partitioner.class", "com.xq.SimplePartitioner");
        ProducerConfig config = new ProducerConfig(props);  
        Producer<String, String> producer = new Producer<String, String>(config);  
        String ip = "192.168.2.3"; 
        String msg ="this is a messageuuu!";
        KeyedMessage<String, String> data = new KeyedMessage<String, String>("test", ip,msg);  
        producer.send(data);
        producer.close();  
    }  
 
}

新建一个消费者例子

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.Message;
import kafka.message.MessageAndMetadata;
 
 
public class ConsumerSample {
 
	public static void main(String[] args) {  
	    // specify some consumer properties  
		Properties props = new Properties();  
		props.put("zookeeper.connect", "10.103.22.47:2181");  
		props.put("zookeeper.connectiontimeout.ms", "1000000");  
		props.put("group.id", "test_group");  
 
	        // Create the connection to the cluster  
		ConsumerConfig consumerConfig = new ConsumerConfig(props);  
		ConsumerConnector connector = Consumer.createJavaConsumerConnector(consumerConfig);  
 
 
	    Map<String,Integer> topics = new HashMap<String,Integer>();  
	    topics.put("test", 2);  
	    Map<String, List<KafkaStream<byte[], byte[]>>> topicMessageStreams = connector.createMessageStreams(topics);  
	    List<KafkaStream<byte[], byte[]>> streams = topicMessageStreams.get("test");
	    ExecutorService threadPool = Executors.newFixedThreadPool(2);  
	    for (final KafkaStream<byte[], byte[]> stream : streams) {  
	    	threadPool.submit(new Runnable() {  
                public void run() {  
                    for (MessageAndMetadata msgAndMetadata : stream) {  
                        // process message (msgAndMetadata.message())  
                        System.out.println("topic: " + msgAndMetadata.topic());  
                        Message message = (Message) msgAndMetadata.message();  
                        ByteBuffer buffer = message.payload();  
                        byte[] bytes = new byte[message.payloadSize()];  
                        buffer.get(bytes);  
                        String tmp = new String(bytes);  
                        System.out.println("message content: " + tmp);  
                    }  
                }  
            });  
        }    
	}
}

先启动消费者例子,然后再启动生产者例子,这样会立即看到效果。

分享到:
评论

相关推荐

    kafka搭建单机windows_单机linux_集群linux操作.rar

    本文将详细介绍如何在Windows单机环境、Linux单机环境以及Linux集群环境下搭建Kafka,旨在帮助读者深入理解Kafka的部署与配置,以便更好地运用在实际项目中。 ### Windows单机环境搭建 1. **下载安装Java运行环境...

    介绍kafka及kafka集群安装

    #### Kafka 单机版部署示例 1. **配置 server.properties 文件**: - 修改 broker.id=0 - 设置 listeners=localhost:9092 - 设置 zookeeper.connect=localhost:2181 2. **启动 ZooKeeper**:执行 `bin/...

    [转]Kafka深度解析

    【标题】:“Kafka深度解析” 【描述】:在信息技术高速发展的今天,分布式消息系统扮演着至关...在“demo-kafka”这个压缩包中,可能包含了Kafka的示例代码或者配置文件,这对于学习和实践Kafka操作具有极大的帮助。

    Kafka的最简安装

    Kafka最简安装方式说明了在单机环境下如何快速搭建起Kafka服务。以下是一些重要的知识点,涵盖了Kafka安装、配置以及基本操作的各个方面: 1. Kafka的版本选择:在本指南中所使用的Kafka版本为kafka_2.10-*.*.*.*。...

    zeromq网络开发库

    ZeroMQ的设计目标是简化分布式系统中的通信问题,使得开发者可以像在单机环境中一样简单地进行多进程通信。这个压缩包文件"zeromq-3.2.3"可能是ZeroMQ的3.2.3版本源码或二进制库,包含了用于构建和使用ZeroMQ的必要...

    面向Java开发人员的Vertx 3指南.zip

    Vert.x是一个轻量级、高性能且多语言的工具包,它允许开发人员构建高度并发、非阻塞的网络应用程序,特别适合微服务和云原生环境。 【描述】"面向Java开发人员的Vertx 3指南" 涵盖了从基础到高级的各个方面,包括...

    raft-multi.zip

    在v2.3.1版本中,Fabric引入了Raft共识算法作为替代原先默认的Kafka,这使得部署更简单,特别是对于单机环境中的小型集群。本文将深入探讨如何构建一个由三个Orderer节点组成的单机Raft集群,以及包含两个组织和三个...

    windows环境Canal总结

    Canal是一款由阿里巴巴开发的开源项目,主要用于MySQL数据库的增量日志解析,并提供增量数据的订阅与消费功能。其核心功能在于能够帮助开发者轻松地捕获数据库中的变化,并将这些变化以一种结构化的形式输出,便于...

    MQ学习文档 方便回顾!

    | 开发语言 | Erlang | Java | Java | Scala & Java | | 协议支持 | AMQP等 | OpenWire等 | 自定义 | 自定义 | | 可用性 | 高 | 一般 | 高 | 高 | | 单机吞吐量 | 一般 | 差 | 高 | 非常高 | | 消息延迟 | 微秒级 | ...

    zookeeper搭建资料及相关组件

    2. **伪集群模式**:在同一台机器上启动多个Zookeeper实例,模拟集群环境,适合开发环境。 3. **集群模式**:在多台机器上部署多个Zookeeper实例,形成真正的集群,提供高可用性和数据一致性。 **三、Zookeeper的...

    storm源码包 apache-0.9.4

    Storm 提供了本地模式(Local Mode)进行开发和测试,可以在单机上模拟完整的集群环境。此外,`storm-starter` 项目包含了示例代码,帮助开发者快速了解如何创建和提交拓扑。 8. **监控与调试**: 通过 Storm UI...

    flink-1.7超详细中文教程

    教程部分将引导学习者从零开始熟悉Flink的环境搭建、基本操作以及编程模型。包括: 1. **安装与配置**:指导如何下载、构建和运行Flink。 2. **编写第一个Flink程序**:通过简单的Word Count示例介绍DataStream ...

    flink-1.10.0-bin-scala_2.11.tgz

    Apache Flink是一个开源流处理和批处理框架,它在大数据处理领域扮演着重要角色。...解压后,用户可以根据官方文档配置环境,启动单机或集群模式的Flink服务,从而开始探索和使用Flink的强大功能。

    聊天服务器.e.rar

    从给定的文件名"聊天服务器.e.rar"来看,这可能是一个关于聊天服务器实现的工程或者代码示例。下面将详细讨论聊天服务器背后涉及的技术和知识点。 1. **网络编程**:聊天服务器的基础是网络编程,通常使用TCP或UDP...

    crawler:分布式爬虫系统

    传统的单机爬虫在面对海量网页时可能会面临效率低下、资源限制等问题。分布式爬虫则通过将爬取任务分散到多台计算机上,实现了并行处理,提高了爬取速度和数据处理能力。它通常由多个部分组成:分布式任务调度、爬虫...

    Flink_Tutorial

    本地模式适合开发调试,独立模式适用于单机部署,而YARN模式则可以在Hadoop集群上运行。 ## 六、Flink容错机制 Flink采用了检查点和保存点的容错策略,保证了处理的exactly-once语义。通过定期保存作业的状态,当...

    spark:spark学习笔记

    5.3 实时流处理示例:通过Spark Streaming处理Kafka数据流,实现实时监控和告警功能。 6.1 性能调优:讨论Spark的内存管理、并行度设置、数据本地性等调优策略,以提高系统性能。 7.1 安装与部署:涵盖Spark的单机...

    golang-crawler:从简单到并发到分布式爬虫

    - 可以使用消息队列(如RabbitMQ、Kafka)进行任务分发,使用数据库或文件系统存储结果,确保数据一致性。 7. **golang-crawler-master项目** - 这个项目的源代码应该包含了实现上述概念的示例代码,包括基本爬虫...

    BD_Arquitecture

    从提供的压缩包文件名"BD_Arquitecture-master"来看,可能包含了一个项目或教程的源代码、文档或示例,这些资源可以帮助深入理解和实践大数据架构。通过学习和应用这些内容,可以提升在大数据领域的技能和知识。

Global site tag (gtag.js) - Google Analytics