背景：

kafka版本：kafka_2.10-0.8.2.1

服务器IP：10.243.3.17

一：Kafkaserver.properties 文件配置

二：zookeeper.properties 文件配置

三： zookeeper,kafka启动

../bin/zookeeper-server-start.sh -daemon /usr/local/kafka_2.10-0.8.2.1/config/zookeeper.properties

../bin/kafka-server-start.sh -daemon /usr/local/kafka_2.10-0.8.2.1/config/server.properties &

四：创建Topic

../bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

五： log4j.xml文件配置

六：常见问题

如遇到问题，首先确定参数配置是否正确，尤其是host,port,advertised.host.name;  然后删除kafka-logs-1,zookeeper-logs; 重新启动zookeeper,kafka;

重新创建topic

七：附录, 非log4j   java连接kafka配置参考

import org.apache.log4j.Logger;

import scala.App;

/**

* TODO:

*

* @author gengchong

* @date 2016年1月5日 上午9:21:16

*/

public class KafkaApp {

private static final Logger LOGGER = Logger.getLogger(App.class);

public static void main(String[] args) throws InterruptedException {

for (int i = 0; i < 20; i++) {

LOGGER.info("Info [" + i + "]");

Thread.sleep(1000);

}

}

}

import java.util.ArrayList;

import java.util.List;

import java.util.Properties;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

/**

* TODO:

*

* @author gengchong

* @date 2016年1月5日 下午1:55:56

*/

public class KafakProducer {

private static final String TOPIC = "test";

private static final String CONTENT = "This is a single message";

private static final String BROKER_LIST = "10.243.3.17:8457";

private static final String SERIALIZER_CLASS = "kafka.serializer.StringEncoder";

public static void main(String[] args) {

Properties props = new Properties();

props.put("serializer.class", SERIALIZER_CLASS);

props.put("metadata.broker.list", BROKER_LIST);

ProducerConfig config = new ProducerConfig(props);

Producer producer = new Producer(config);

//Send one message.

KeyedMessage message =

new KeyedMessage(TOPIC, CONTENT);

producer.send(message);

//Send multiple messages.

List messages =

new ArrayList();

for (int i = 0; i < 5; i++) {

messages.add(new KeyedMessage

(TOPIC, "============== send Message. " + i));

}

producer.send(messages);

}

}

import java.util.List;

import java.util.Map;

import java.util.Properties;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import com.google.common.collect.ImmutableMap;

import kafka.consumer.Consumer;

import kafka.consumer.ConsumerConfig;

import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

import kafka.message.MessageAndMetadata;

/**

* TODO:

*

* @author gengchong

* @date 2016年1月5日 上午9:22:04

*/

public class KafkaConsumer {

private static final String ZOOKEEPER = "10.243.3.17:2181";

//groupName可以随意给，因为对于kafka里的每条消息，每个group都会完整的处理一遍

private static final String GROUP_NAME = "test_group";

private static final String TOPIC_NAME = "test";

private static final int CONSUMER_NUM = 4;

private static final int PARTITION_NUM = 4;

public static void main(String[] args) {

// specify some consumer properties

Properties props = new Properties();

props.put("zookeeper.connect", ZOOKEEPER);

props.put("zookeeper.connectiontimeout.ms", "1000000");

props.put("group.id", GROUP_NAME);

// Create the connection to the cluster

ConsumerConfig consumerConfig = new ConsumerConfig(props);

ConsumerConnector consumerConnector =

Consumer.createJavaConsumerConnector(consumerConfig);

// create 4 partitions of the stream for topic “test”, to allow 4

// threads to consume

Map> topicMessageStreams =

consumerConnector.createMessageStreams(

ImmutableMap.of(TOPIC_NAME, PARTITION_NUM));

List streams = topicMessageStreams.get(TOPIC_NAME);

// create list of 4 threads to consume from each of the partitions

ExecutorService executor = Executors.newFixedThreadPool(CONSUMER_NUM);

// consume the messages in the threads

for (final KafkaStream stream : streams) {

executor.submit(new Runnable() {

public void run() {

for (MessageAndMetadata msgAndMetadata : stream) {

// process message (msgAndMetadata.message())

System.out.println(new String(msgAndMetadata.message()));

}

}

});

}

}

}